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", ""))
# 0.9.0 (AR5): busy=<async () -> int> reports running work an update would break; the engine waits while > 0

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,
    has_second_factor=lambda u: bool(u.get("mfa_enabled")), verify_second_factor=check_mfa_code,
    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, second_factor} (SaaS: not available, the customer portal resets it); POST takes {confirm: "RESET", password, code, remove_license}. Since 0.11.0 (N21, security review 2026-10-27) the second factor is required: has_second_factor(user) and verify_second_factor(user, code) are mandatory, verify_second_factor checks a TOTP or backup code and consumes it (last step stored, backup code removed). An admin without a second factor gets second_factor_missing, a wrong code second_factor_wrong; the password is checked as well when the account has one. allowed(user) narrows who may (Foundry: platform owner), error(code) maps the codes reset_in_portal, reset_confirm_wrong, current_password_wrong, second_factor_missing, second_factor_wrong to the app's own errors. The frontend is FactoryReset from @scalio-cloud/ui (0.32.0 with the code field).

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.

Since 0.10.0: queue(to, subject, body, ref=...) puts a free mail into the same outbox (own subject and text, plain text or HTML, no template), for mails whose text the customer writes elsewhere (Nexus workflow steps); it returns False like send and runs check. An app with its own connection to the customer's tenant passes graph_token= (async, returns a Graph token or raises MailError): the settings then offer the provider tenant («Microsoft 365 über die Tenant-Verbindung», scalio-ui 0.31.0), which needs only the sending mailbox graph.sender and no second app secret. settings() lists the offered providers in providers.