# `sandloom.sdk.infra` Infrastructure accessors, settings, the core runtime and (advanced) the facade. Public API 1.0 (spec §2.2, Appendix C). `from sandloom.sdk.infra import `. Signatures in a terminal: `sandloom docs `. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line `# sandloom: advanced ` in generated code and examples). ## Advanced | Name | Kind | Summary | | --- | --- | --- | | [`Cache`](<#cache>) | class | What a module calls on `cache()`: the operations, not the lifecycle. | | [`cache`](<#cache_1>) | function | Return the operation-based `CacheProvider` for a named instance. | | [`capability`](<#capability>) | function | Return the instance `name` of the capability `ref` points to, typed by the ref. A capability's package publishes the ref and the caller's... | | [`CapabilityRef`](<#capabilityref>) | class | A typed key to a capability: `CapabilityRef[Search]("search.v1")`. | | [`claims_contract_name`](<#claims_contract_name>) | function | `"tenancy"` -\> `"tenancy.claims.v1"`. | | [`CLAIMS_CONTRACT_SUFFIX`](<#claims_contract_suffix>) | constant | | | [`ClaimsContributorContract`](<#claimscontributorcontract>) | class | `.claims.v1` — a module's token-claims contributors as a contract. | | [`core_from_request`](<#core_from_request>) | function | SYNC accessor of the request's `CoreRuntime` (direct calls). | | [`CoreFacade`](<#corefacade>) | class | A module's strictly bounded access point to the framework's infrastructure. | | [`CoreRuntime`](<#coreruntime>) | class | Read-only runtime view of the `CoreFacade` for module code. | | [`facts_contract_name`](<#facts_contract_name>) | function | `"billing"` -\> `"billing.facts.v1"`. | | [`FACTS_CONTRACT_SUFFIX`](<#facts_contract_suffix>) | constant | | | [`get_core`](<#get_core>) | async function | Return the read-only `CoreRuntime` view of the app's facade. | | [`Notification`](<#notification>) | class | One outbound notification, independent of the delivery backend. | | [`Notifier`](<#notifier>) | class | What a module calls on `notifier()`. | | [`notifier`](<#notifier_1>) | function | Return the `NotifierProvider` for a named instance. | | [`ObjectStore`](<#objectstore>) | class | What a module calls on `objectstore()`. | | [`objectstore`](<#objectstore_1>) | function | Return the `ObjectStoreProvider` for a named instance (the module-level twin of `CoreRuntime.objectstore`; spec §2.2). | | [`settings`](<#settings>) | settings object | | | [`ShutdownTask`](<#shutdowntask>) | constant | | | [`StartupTask`](<#startuptask>) | constant | | | [`TaskQueue`](<#taskqueue>) | class | What a module calls on `taskqueue()`. | | [`taskqueue`](<#taskqueue_1>) | function | Return the `TaskQueueProvider` for a named instance (the module-level twin of `CoreRuntime.taskqueue`; spec §2.2). | ## Advanced names ### `Cache` ``` class Cache(Protocol) ``` What a module calls on `cache()`: the operations, not the lifecycle. #### `Cache.get` ``` async def get(self, key: str) -> str | None ``` #### `Cache.set` ``` async def set(self, key: str, value: str, *, ttl_seconds: int | None = None) -> None ``` #### `Cache.add` ``` async def add(self, key: str, value: str, *, ttl_seconds: int | None = None) -> bool ``` #### `Cache.delete` ``` async def delete(self, key: str) -> None ``` #### `Cache.incr` ``` async def incr(self, key: str, *, ttl_seconds: int | None = None) -> int ``` #### `Cache.get_many` ``` async def get_many(self, keys: Sequence[str]) -> list[str | None] ``` #### `Cache.set_many` ``` async def set_many( self, items: Mapping[str, str], *, ttl_seconds: int | None = None, ) -> None ``` #### `Cache.compare_and_set` ``` async def compare_and_set( self, key: str, expected: str, value: str, *, ttl_seconds: int | None = None, ) -> bool ``` #### `Cache.compare_and_delete` ``` async def compare_and_delete(self, key: str, expected: str) -> bool ``` #### `Cache.incr_window` ``` async def incr_window( self, key: str, previous_key: str, *, ttl_seconds: int | None = None, ) -> tuple[int, str | None] ``` #### `Cache.incr_many` ``` async def incr_many( self, keys: Sequence[str], *, ttl_seconds: int | None = None, ) -> list[int] ``` #### `Cache.set_nx_get` ``` async def set_nx_get( self, key: str, value: str, *, ttl_seconds: int | None = None, ) -> tuple[bool, str | None] ``` #### `Cache.set_max` ``` async def set_max(self, key: str, value: str, *, ttl_seconds: int | None = None) -> str ``` ### `cache` ``` def cache(name: str = 'default') -> Cache ``` Return the operation-based `CacheProvider` for a named instance. The sanctioned cache surface for modules and core guards — `get`/`set`/ `add`/`delete`/`incr` (see `sandloom.core.infra.contracts.CacheProvider`). Resolves via the process-active `InfraRegistry` (`RuntimeError` when none is bound); the registry's configuration picks the driver, so this surface never names a concrete backend. ### `capability` ``` def capability(ref: CapabilityRef[P], name: str = 'default') -> P ``` Return the instance `name` of the capability `ref` points to, typed by the ref. A capability's package publishes the ref and the caller's Protocol, and the capability's major version must match the ref's. ### `CapabilityRef` ``` class CapabilityRef(Generic)(value: str) -> None ``` A typed key to a capability: `CapabilityRef[Search]("search.v1")`. The caller's Protocol travels in the type, and a `.vN` suffix is checked against the capability when it is resolved; a ref without one skips that check. A class passed as `type[T]` cannot do this: mypy refuses an abstract class or a Protocol there. #### `CapabilityRef.name` ``` name (property) ``` #### `CapabilityRef.version` ``` version (property) ``` The `vN` suffix, if any. Only a versioned ref is checked on resolve. ### `claims_contract_name` ``` def claims_contract_name(owner: str) -> str ``` `"tenancy"` -\> `"tenancy.claims.v1"`. ### `CLAIMS_CONTRACT_SUFFIX` ``` CLAIMS_CONTRACT_SUFFIX: str = '.claims.v1' ``` *(no docstring)* ### `ClaimsContributorContract` ``` class ClaimsContributorContract(facade: CoreFacade, owner: str | None) -> None ``` `.claims.v1` — a module's token-claims contributors as a contract. Auto-registered for every module that adds a claims contributor; runs the module's contributors (in its own read-only Unit of Work, like any read contract) and returns their merged claims. A token issuer in another process reaches it over internal RPC via `SERVICE_BINDINGS`. #### `ClaimsContributorContract.contribute` ``` async def contribute(self, user_id: Any) -> dict[str, Any] ``` Merged claims of this module's contributors for `user_id`. ### `core_from_request` ``` def core_from_request(request: Request) -> CoreRuntime ``` SYNC accessor of the request's `CoreRuntime` (direct calls). For code that holds a `Request` and cannot `await`: the same value as `await get_core(request)`. As a dependency use `Depends(get_core)` (async: no thread-pool hop), never `Depends(core_from_request)`. ### `CoreFacade` ``` class CoreFacade(_ClaimsMixin, _RegistriesMixin)( app: FastAPI, settings_obj: Settings | None = None, service_registry: ServiceRegistry | None = None, get_db: Callable[[], Awaitable[AsyncSession]] | Callable[[], AsyncGenerator[AsyncSession, None]] | None = None, ) -> None ``` A module's strictly bounded access point to the framework's infrastructure. Provides only methods that are safe and stable for modules to depend on; concrete infrastructure details (cache backend, tracing implementation, transport) stay hidden behind this API. One instance is created by the composition root and handed to every module's `register(core)`. #### `CoreFacade.event_bus` ``` event_bus (property) ``` Return the event bus owned by this facade. The instance is bound via set\_event\_bus() during app wiring so each application owns (and shuts down) exactly its own bus. Falls back to the bare-facade in-memory bus for facades that never bind one (unit tests). #### `CoreFacade.set_event_bus` ``` def set_event_bus(self, bus: Any) -> None ``` Bind the event bus instance this application owns. #### `CoreFacade.add_startup_task` ``` def add_startup_task( self, task: StartupTask, *, critical: bool = False, phase: str | None = None, ) -> None ``` Register a startup task. critical=True means the application MUST NOT start if the task fails: run\_startup re-raises instead of degrading silently. `phase` is `"infra"` (runs first), `"modules"` (the default) or `"post"` (runs last); a task's name never selects its phase. #### `CoreFacade.is_critical_startup_task` ``` def is_critical_startup_task(self, task: Any) -> bool ``` #### `CoreFacade.startup_task_specs` ``` startup_task_specs (property) ``` `(task, critical)` pairs in registration order. #### `CoreFacade.startup_task_phased_specs` ``` startup_task_phased_specs (property) ``` `(task, critical, phase)` triples in registration order. #### `CoreFacade.add_shutdown_task` ``` def add_shutdown_task(self, task: ShutdownTask) -> None ``` #### `CoreFacade.register_health_check` ``` def register_health_check(self, name: str, fn: HealthCheck) -> None ``` #### `CoreFacade.health_checks` ``` health_checks (property) ``` #### `CoreFacade.startup_tasks` ``` startup_tasks (property) ``` #### `CoreFacade.shutdown_tasks` ``` shutdown_tasks (property) ``` #### `CoreFacade.module_registration` ``` def module_registration(self, module_name: str) -> Iterator[ModuleAPI] ``` Scope a module registration so a failure leaves no partial state. Yields a draft-backed `ModuleAPI`: everything the module registers (routers, services, lifecycle hooks, health checks, event schemas) is buffered in a `ModuleRegistrationDraft` and applied atomically by `commit` only when `register()` returns normally. A failure propagates without ever committing, so NOTHING is applied — including event schemas, which the old snapshot/restore path left behind as orphans in the contract inventory. This replaces snapshotting private kernel state (`service_registry._map`, `app.router.routes` slicing): the kernel is never mutated until commit. #### `CoreFacade.ensure_tracer` ``` def ensure_tracer(self) -> None ``` Initialize tracing once (idempotent). #### `CoreFacade.span` ``` def span(self, name: str) -> Any ``` Return a span context manager (no-op if tracing disabled). Usage: with core.span("my.operation"): ... #### `CoreFacade.register_metric` ``` def register_metric(self, name: str, value: Any | None = None) -> None ``` #### `CoreFacade.increment_metric` ``` def increment_metric(self, name: str, amount: int | float = 1) -> None ``` Atomically add `amount` to the counter `name`. A real counter (monotonic, thread-safe): the in-process value (`get_metric`) is updated under a lock and the increment is exported to the Prometheus counter `sandloom_facade_counter_total{name=...}`. Use it instead of a `get_metric`/`register_metric` read-modify-write. #### `CoreFacade.get_metric` ``` def get_metric(self, name: str) -> Any ``` #### `CoreFacade.list_metrics` ``` def list_metrics(self) -> dict[str, Any] ``` #### `CoreFacade.logger_factory` ``` def logger_factory(self, name: str) -> logging.Logger ``` Return namespaced module logger (sandloom.module.\). ### `CoreRuntime` ``` class CoreRuntime(facade: CoreFacade) -> None ``` Read-only runtime view of the `CoreFacade` for module code. Request-time module code (`Depends(get_core)`) gets THIS, not the full facade: it can resolve contracts and capabilities, read settings, log, trace, count, and collect token claims — but it cannot reach the event bus (publish through `publish_event`), the infrastructure registry, the raw FastAPI app, the service registry, or any `register_*` mutator. Registration happens only through the `sandloom.core.module_api.ModuleAPI` handed to `register(core)`. This is an API-discipline boundary for trusted application code, not an immutable object graph or a sandbox. Shared Settings must be treated as read-only by modules; the remote-fact cache is maintained by shared authz machinery. Do not mutate either from a business module. #### `CoreRuntime.get_service` ``` def get_service(self, name: str) -> Any ``` #### `CoreRuntime.collect_claims` ``` async def collect_claims( self, user_id: Any, issuer_session: Any = None, *, issuer: str | None = None, precomputed: Mapping[Any, dict[str, Any] | None] | None = None, ) -> dict[str, Any] ``` #### `CoreRuntime.get_access_fact` ``` def get_access_fact(self, name: str) -> Any ``` #### `CoreRuntime.remote_fact_owners` ``` def remote_fact_owners(self) -> list[str] ``` #### `CoreRuntime.remote_access_fact_cache` ``` remote_access_fact_cache (property) ``` #### `CoreRuntime.capability` ``` def capability( self, capability: str, name: str = 'default', contract: type | None = None, ) -> Any ``` #### `CoreRuntime.require_capability` ``` def require_capability( self, capability: str, name: str = 'default', *, features: set[str] | None = None, contract: type | None = None, version: str | None = None, ) -> Any ``` #### `CoreRuntime.database` ``` def database(self, name: str = 'default') -> Any ``` #### `CoreRuntime.cache` ``` def cache(self, name: str = 'default') -> Any ``` #### `CoreRuntime.objectstore` ``` def objectstore(self, name: str = 'default') -> Any ``` #### `CoreRuntime.taskqueue` ``` def taskqueue(self, name: str = 'default') -> Any ``` #### `CoreRuntime.notifier` ``` def notifier(self, name: str = 'default') -> Any ``` #### `CoreRuntime.settings` ``` settings (property) ``` #### `CoreRuntime.logger` ``` logger (property) ``` #### `CoreRuntime.logger_factory` ``` def logger_factory(self, name: str) -> logging.Logger ``` #### `CoreRuntime.span` ``` def span(self, name: str) -> Any ``` #### `CoreRuntime.increment_metric` ``` def increment_metric(self, name: str, amount: int | float = 1) -> None ``` #### `CoreRuntime.get_metric` ``` def get_metric(self, name: str) -> Any ``` ### `facts_contract_name` ``` def facts_contract_name(owner: str) -> str ``` `"billing"` -\> `"billing.facts.v1"`. ### `FACTS_CONTRACT_SUFFIX` ``` FACTS_CONTRACT_SUFFIX: str = '.facts.v1' ``` *(no docstring)* ### `get_core` ``` async def get_core(request: Request) -> CoreRuntime ``` Return the read-only `CoreRuntime` view of the app's facade. Module code gets contract/capability resolution, settings, logging, tracing and counters — never the event bus, the infra registry, the raw app or the `register_*` mutators (those belong to `register(core)`). `async` on purpose: it does no I/O, and FastAPI runs a SYNC dependency in the anyio thread pool on every request (a thread hop for an attribute read). Use it as `Depends(get_core)`; a direct call is `core = await get_core(request)`. ### `Notification` ``` class Notification( channel: NotificationChannel, recipient: str, subject: str, body: str, data: Mapping[str, Any] = , dedupe_key: str | None = None, body_fields: Mapping[str, str] = , ) -> None ``` One outbound notification, independent of the delivery backend. A logical message: which `channel` to deliver on, the `recipient` address for that channel (an email, phone, device token, or user id for `inapp`), a short `subject` and `body`, plus structured `data` and an optional `dedupe_key` a driver may use to suppress duplicate sends. `body_fields` are the values the `body` was built from, so a driver can render its own template. They are as sensitive as the body (a secret link, a name): a driver never logs them, and `data` stays safe to record. ### `Notifier` ``` class Notifier(Protocol) ``` What a module calls on `notifier()`. #### `Notifier.send` ``` async def send(self, notification: Notification) -> None ``` ### `notifier` ``` def notifier(name: str = 'default') -> Notifier ``` Return the `NotifierProvider` for a named instance. The sanctioned outbound-notification surface for modules: build a `sandloom.core.infra.contracts.Notification` and `await notifier().send(...)` — never import an email/SMS SDK. Resolves via the process-active `InfraRegistry` (`RuntimeError` when none is bound). ### `ObjectStore` ``` class ObjectStore(Protocol) ``` What a module calls on `objectstore()`. #### `ObjectStore.put` ``` async def put( self, key: str, data: bytes | Any, *, content_type: str | None = None, metadata: Mapping[str, str] | None = None, ) -> str ``` #### `ObjectStore.get` ``` async def get(self, key: str) -> bytes ``` #### `ObjectStore.delete` ``` async def delete(self, key: str) -> None ``` #### `ObjectStore.url` ``` async def url(self, key: str) -> str ``` ### `objectstore` ``` def objectstore(name: str = 'default') -> ObjectStore ``` Return the `ObjectStoreProvider` for a named instance (the module-level twin of `CoreRuntime.objectstore`; spec §2.2). ### `settings` ``` settings: Settings # loaded and validated at first attribute access ``` *(no docstring)* ### `ShutdownTask` ``` ShutdownTask: UnionType ``` *(no docstring)* ### `StartupTask` ``` StartupTask: UnionType ``` *(no docstring)* ### `TaskQueue` ``` class TaskQueue(Protocol) ``` What a module calls on `taskqueue()`. #### `TaskQueue.enqueue` ``` async def enqueue( self, task: str, payload: Mapping[str, Any], *, delay_seconds: int | None = None, idempotency_key: str | None = None, ) -> str ``` ### `taskqueue` ``` def taskqueue(name: str = 'default') -> TaskQueue ``` Return the `TaskQueueProvider` for a named instance (the module-level twin of `CoreRuntime.taskqueue`; spec §2.2). ## Examples and related guides - [Infrastructure capabilities]() - [Python API overview]()