scalio-platform

Shared building blocks of the Scalio apps (Nexus, Compass, Citadel, Foundry, Rosetta, Pulse, Beacon, Cockpit, ID, Monitor). Before this package each app kept its own copy of these files (architecture review C2/C4, ADR 0058 in scalio-docs). Nothing in this package is secret.

Use in an app

scalio-platform==0.1.0            # backend/requirements.txt, exact version

from scalio_platform import error_report, health, job_lease
from scalio_platform.migrations import BASELINE, run_migrations

MIGRATIONS = BASELINE                      # append (2, migrate_002_...) and so on, never edit past steps

async def startup():
    await run_migrations(db, MIGRATIONS)   # before the background loops
    for name, loop in LOOPS.items():
        asyncio.create_task(job_lease.run_exclusive(db, name, loop))

error_report.install(app, "citadel", build_info)
code, body = await health.health_report(db, build_info(), lease_jobs=sorted(LOOPS))

# probe of the provisioning engine (SaaS on the Docker engine), header X-Provision-Health-Secret
code, body = await health.provision_report(db, build_info(), secret=os.environ.get("PROVISION_HEALTH_SECRET", ""),
                                           provided=request.headers.get("x-provision-health-secret", ""))

A migration step must be safe to run twice: two instances can start at the same moment.

Zurücksetzen (factory reset, ADR 0064 / 0069)

from scalio_platform.factory_reset import factory_reset_router

async def after_wipe(remove_license: bool):
    if remove_license:
        await licensing.runtime.refresh(db)      # free edition right away
    await login_policy.announce_setup_code()     # new setup code in the log

app.include_router(factory_reset_router(
    get_db=lambda: db, current_user=require_admin,
    password_hash=lambda u: u.get("password_hash"), verify_password=auth.verify_password,
    keep_meta=(f"{PRODUCT.code}_installation", "schema_version"), license_meta=(PRODUCT.meta_id,),
    saas_mode=saas_mode, after_wipe=after_wipe))

GET /api/admin/factory-reset answers {available, password} (SaaS: not available, the customer portal resets it); POST takes {confirm: "RESET", password, remove_license}. allowed(user) narrows who may (Foundry: platform owner), error(code) maps the codes reset_in_portal, reset_confirm_wrong, current_password_wrong to the app's own errors. The frontend is FactoryReset from @scalio-cloud/ui 0.27.0.

Develop and release

• pip install -e ".[dev]", then pytest -q and ruff check . (both run in CI).
• Release: bump pyproject.toml and __version__, open a scalio-docs PR with python3 tools/changelog.py --repo scalio-platform release <version> (changelog in scalio-docs changelog/scalio-platform/, ADR 0066), merge, then «Freigeben» in the Scalio Cockpit (workflow publish.yml: PyPI with trusted publishing, tag and GitHub release). Tags without v.
• A change of behaviour changes every app on its next upgrade: say so in the PR and the CHANGELOG.

Protokoll: Aktionen und Jobs (ADR 0070)

from scalio_platform import audit, jobs
from scalio_platform.audit import Category

CATEGORIES = (Category("access", "Zugriff", "Access", ("user.", "role.")),
              Category("license", "Lizenz", "License", ("license.",), always=True))

await audit.record(db, "user.create", user, category="access", target=audit.target("user", u["id"], u["email"]),
                   detail=f"Konto angelegt · {u['email']}", request=request)

app.include_router(audit.audit_router(get_db=lambda: db, viewer=require_perm(CATALOG, me, get_db, "audit.view"),
                                      categories=CATEGORIES, tenant_scope=lambda u: None), prefix="/api")

# worker
REGISTRY = jobs.Registry([jobs.Job("mail_intake", "Mail-Eingang", "Mail intake", "Jede Minute", "Every minute")])
await jobs.run(db, "mail_intake", mail_intake)            # inside the loop; return None when nothing happened
await jobs.serve_requests(db, REGISTRY, HANDLERS)         # own loop every few seconds («Jetzt ausführen»)

# API
app.include_router(jobs.jobs_router(registry=REGISTRY, get_db=lambda: db, viewer=view, runner=run_perm,
                                    queue=MyQueue(), audit=log), prefix="/api")

Call audit.ensure_indexes(db) and jobs.ensure_indexes(db) at startup. Never write passwords, tokens or contents into detail, changes or a job summary (changes drops fields named like pass/secret/token/key).

E-Mail (ADR 0069 W5)

from scalio_platform import mail

KINDS = (mail.MailKind("resolved", ("Ticket erledigt", "Ticket done"),
                       ("An die meldende Person, wenn ein Ticket erledigt ist", "To the reporter when a ticket is done"),
                       subject={"de": "[{key}] Erledigt: {title}", "en": "[{key}] Done: {title}"},
                       body={"de": "Guten Tag {name}\n\n...\n\n{sender}", "en": "Hello {name}\n\n...\n\n{sender}"},
                       tokens=("key", "title", "name", "link"), sample={"key": "IT-12", "title": "Drucker", ...},
                       switchable=True),)

mailer = mail.Mailer(get_db=lambda: db, app_name="Scalio Beacon", kinds=KINDS,
                     seal=secret_box.seal, unseal=secret_box.unseal,      # the app's own encryption
                     collection="meta",                                   # default "settings"
                     override=scalio_selftest.mail.smtp_settings,         # test mailbox on test instances
                     allowed=scalio_selftest.mail.allowed, base_url=lambda: os.environ.get("APP_PUBLIC_URL", ""))
app.include_router(mailer.router(admin=require_admin, audit=log), prefix="/api")   # /api/mail/...

await mailer.send(person["email"], "resolved", {"key": t["key"], ...}, lang=person.get("language"), ref=t["id"])
await jobs.run(db, "mail_outbox", mailer.flush)        # worker, every 30 seconds; call mailer.ensure_indexes() once

Every template knows {app_name}, {app_url} and {sender} (sender name of the settings, else the app name). A template may be plain text or HTML (from the editor); in HTML the values are escaped. The built-in kind test is the test mail. send returns False when mail is off, the kind is switched off, or allowed refuses the address. check(subject, html) may refuse a mail with ValueError (Foundry: no forms, only https links); it runs on saving or testing a template and before a mail goes into the outbox.