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 <name>. Signatures in a terminal: sandloom docs <name>. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line # sandloom: advanced <reason> in generated code and examples).

Advanced¶

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

<module>.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.<name>).

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] = <factory>,
    dedupe_key: str | None = None,
    body_fields: Mapping[str, str] = <factory>,
) -> 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