# `sandloom.sdk.module` Modules: `define_module`, the L2 `register` form (`ModuleAPI`, `ModuleDescriptor`), requirements, health and loader results. Public API 1.0 (spec §2.2, Appendix C). `from sandloom.sdk.module 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). ## Start with (beginner) | Name | Kind | Summary | | --- | --- | --- | | [`define_module`](<#define_module>) | function | Build a module's `register(core)` from its declarations (spec §5.7). | ## Advanced | Name | Kind | Summary | | --- | --- | --- | | [`AUTO`](<#auto>) | constant | | | [`CapabilityRequirement`](<#capabilityrequirement>) | class | An infrastructure capability a module needs to function. | | [`ContractRequirement`](<#contractrequirement>) | class | A sync service contract a module CALLS (`core.get_service(name)`). | | [`FailedModule`](<#failedmodule>) | class | A module that failed to load, with its error and whether it was required. | | [`HealthCheck`](<#healthcheck>) | callable | | | [`HealthResult`](<#healthresult>) | class | Outcome of a module health check: a status plus optional details. | | [`HealthStatus`](<#healthstatus>) | callable | | | [`MigrationTargetError`](<#migrationtargeterror>) | exception | A module's migration target database cannot be determined unambiguously. | | [`ModuleAPI`](<#moduleapi>) | class | Stable, minimal registration surface for business modules. | | [`ModuleDescriptor`](<#moduledescriptor>) | class | What a module returns from `register(core)` to describe itself. | | [`ModuleLoadSummary`](<#moduleloadsummary>) | class | Aggregate result of a load pass: loaded, skipped, failed, and duplicates. | | [`SkippedModule`](<#skippedmodule>) | class | A module the loader skipped, with the reason why. | ## Beginner names ### `define_module` ``` def define_module( package: str | None = None, /, *, version: str = '0.1.0', optional: bool = False, consumes: Sequence[Any] = (), capabilities: Sequence[Any] = (), depends_on: Sequence[str] = (), migrations_database: str | None = None, migrations_chain: str | None = None, health_check: Any | None = None, setup: Callable[[ModuleAPI], None] | None = None, name: str | None = None, router: Any = 'auto', events: Any = 'auto', issues_tokens: bool = False, ) -> Callable[[ModuleAPI], Any] ``` Build a module's `register(core)` from its declarations (spec §5.7). In the module's `__init__.py`: ``` from sandloom.sdk import define_module register = define_module(__name__, version="0.1.0") ``` Everything the module declares in its conventional submodules is discovered (routers, events, consumers, contracts, sagas, projections, consumed contracts). `setup(core)` stays for what has no declaration (startup tasks, claims contributors, planner operations). `router=` / `events=` override discovery as today; `consumes` adds to the derived requirements. Modules are REQUIRED unless `optional=True`. ## Advanced names ### `AUTO` ``` AUTO: str = 'auto' ``` *(no docstring)* ### `CapabilityRequirement` ``` class CapabilityRequirement( capability: str, instance: str = 'default', version: str = 'v1', features: frozenset[str] = , required: bool = True, ) -> None ``` An infrastructure capability a module needs to function. Declared on the `ModuleDescriptor` so the loader can preflight — at startup, before serving traffic — that the capability exists, has a configured instance backed by a registered driver, and that the driver supports every required feature. A `required` requirement that cannot be satisfied fails the module load (and aborts startup if the module itself is required); an optional one only logs a warning. ### `ContractRequirement` ``` class ContractRequirement( name: str, required: bool = True, reason: str = '', methods: tuple[str, ...] = (), ) -> None ``` A sync service contract a module CALLS (`core.get_service(name)`). Declared on `ModuleDescriptor.consumes` so the loader can check at startup — not at the first request — that every consumed contract is either provided by a module loaded in this process or bound remotely in `SERVICE_BINDINGS`. A `required` contract that is neither aborts startup (`ContractBindingError`); an optional one (the module degrades explicitly without it) only logs a warning. Access facts owned elsewhere are consumed through the owner's `.facts.v1` contract: `ContractRequirement("billing.facts.v1")`. ### `FailedModule` ``` class FailedModule(name: str, error: str, required: bool) -> None ``` A module that failed to load, with its error and whether it was required. ### `HealthCheck` ``` def HealthCheck(*args, **kwargs) ``` *(no docstring)* ### `HealthResult` ``` class HealthResult( status: HealthStatus, details: dict[str, object] | None = None, ) -> None ``` Outcome of a module health check: a status plus optional details. ### `HealthStatus` ``` def HealthStatus(*args, **kwargs) ``` *(no docstring)* ### `MigrationTargetError` ``` class MigrationTargetError(ValueError) ``` A module's migration target database cannot be determined unambiguously. ### `ModuleAPI` ``` class ModuleAPI( facade: CoreFacade, *, draft: ModuleRegistrationDraft | None = None, ) -> None ``` Stable, minimal registration surface for business modules. When constructed with a `draft` (the loader's transactional path) write methods buffer into it and are applied only on a successful commit; read methods always reflect the live facade. Without a draft (direct use in tests) writes apply to the facade immediately, preserving the simple API. #### `ModuleAPI.include_router` ``` def include_router(self, router: APIRouter, **kwargs: Any) -> None ``` Mount a module's router (replaces reaching into `core.app`). #### `ModuleAPI.register_service` ``` def register_service(self, name: str, impl: Any, *, database: str | None = None) -> None ``` Register a sync service contract provided by this module. `database` overrides the database the contract runtime opens the provider's Unit of Work on; by default it is the database the module's descriptor declares (`ModuleDescriptor.database_instances()`). #### `ModuleAPI.get_service` ``` def get_service(self, name: str) -> Any ``` #### `ModuleAPI.register_event_schema` ``` def register_event_schema( self, channel: str, name: str, schema: dict[str, Any] | None = None, ) -> Any ``` #### `ModuleAPI.register_event_schema_obj` ``` def register_event_schema_obj(self, event_schema: Any) -> Any ``` #### `ModuleAPI.get_schema_registry` ``` def get_schema_registry(self) -> Any ``` #### `ModuleAPI.add_startup_task` ``` def add_startup_task(self, task: StartupTask, *, critical: bool = False) -> None ``` #### `ModuleAPI.add_shutdown_task` ``` def add_shutdown_task(self, task: ShutdownTask) -> None ``` #### `ModuleAPI.register_health_check` ``` def register_health_check(self, name: str, fn: HealthCheck) -> None ``` #### `ModuleAPI.add_claims_contributor` ``` def add_claims_contributor(self, fn: Any, *, critical: bool = False) -> None ``` Register a contributor of JWT claims (`async (user_id) -> dict`). The contributor runs in the module's OWN Unit of Work on its own database (read it with `contract_session`); it is also exposed as the `.claims.v1` contract so a token issuer in another process can collect it over RPC (`SERVICE_BINDINGS`). `critical=True` aborts token minting (503) if the contributor fails (use for security claims like roles/permissions and tenant allowlists). #### `ModuleAPI.register_query_projection` ``` def register_query_projection(self, descriptor: Any) -> None ``` Declare a denormalised cross-module read model this module owns. #### `ModuleAPI.register_projection_consumer` ``` def register_projection_consumer(self, descriptor: Any) -> None ``` Declare a durable consumer that applies a source channel to a projection. #### `ModuleAPI.register_projection_rebuilder` ``` def register_projection_rebuilder(self, projection: str, rebuilder: Any) -> None ``` Register the rebuilder that reconstructs `projection` from source. #### `ModuleAPI.register_event_consumer` ``` def register_event_consumer( self, channel: str, handler: Any, *, name: str, durable: bool = True, database: str = 'default', required: bool = True, group: str | None = None, start_from: str = 'earliest', bind_tenant_from: str | None = None, dead_letter: bool = True, ) -> None ``` Subscribe `handler` to `channel` as a durable consumer. The sanctioned "react to another module's event with a side effect" path: at startup the framework subscribes the handler under its own durable consumer group `:` (namespaced by the registering module, so two modules — or this module deployed in two services — never share a group and split its events) and runs each delivery in a Unit of Work on `database` (commit on success, rollback+redelivery on error). With `durable` (default) the event id is claimed for exactly-once apply. `required=True` (default) gates application readiness on the live subscription; mark deliberately best-effort consumers optional. `group` pins an explicit group id instead (e.g. the bare pre-0.4 name, to keep draining an existing group). `start_from` applies when the group is first created: `"earliest"` (default; every event still within `EVENT_STREAM_RETENTION_SECONDS`), `"latest"` or an ISO-8601 timestamp. `bind_tenant_from` (opt-in) names the payload field holding the tenant id: the handler then runs with that tenant bound, so `TenantScoped` code works; a missing/invalid value fails the delivery. The handler signature is `async (session, envelope) -> None`. `dead_letter=False` retries a failing delivery FOREVER (capped backoff) instead of dead-lettering it after the delivery budget — for effects that must never be silently parked; the handler raises `PoisonEventError` (`sandloom.core.di`) for an event that can never apply, which is dead-lettered at once with a CRITICAL log. #### `ModuleAPI.register_saga` ``` def register_saga(self, definition: Any) -> None ``` Register a saga this module owns (`sandloom.core.module_api.Saga`). Binds it to the module (`saga_type = "."`) and adds a `ContractRequirement(name, methods=...)` for every contract its steps call to the module's `consumes` — a missing provider is then a `ContractBindingError` at startup. The kernel checks the calls against the providers' describe documents once every module loaded and installs the advance consumer `:sagas@advance` and the timer sweep. #### `ModuleAPI.register_access_fact` ``` def register_access_fact(self, fact: Any) -> None ``` Register an `sandloom.shared.authz.facts.AccessFact`. Binds a fact name to its claim/projection/live resolution paths so `Entitled(name)` policies resolve it by configuration. #### `ModuleAPI.register_operations` ``` def register_operations(self, *operations: Any) -> None ``` Declare this module's planner operations (`ReadOperation` / `JoinRead`): each is planned at startup (a `critical` startup task, so a declaration that does not match the schema fails the boot), the chosen executor and the reason are logged, and `/diagnostics/core` lists them (`planner`). #### `ModuleAPI.logger_factory` ``` def logger_factory(self, name: str) -> logging.Logger ``` #### `ModuleAPI.logger` ``` logger (property) ``` #### `ModuleAPI.settings` ``` settings (property) ``` ### `ModuleDescriptor` ``` class ModuleDescriptor( name: str, version: str, optional: bool = False, routes: APIRouter | None = None, health_check: HealthCheck | None = None, migrations_path: str | None = None, depends_on: list[str] = , capabilities: list[CapabilityRequirement] = , migrations_database: str | None = None, migrations_chain: str | None = None, consumes: list[ContractRequirement] = , issues_tokens: bool = False, ) -> None ``` What a module returns from `register(core)` to describe itself. Carries the module's identity (`name`/`version`), whether it is `optional` (modules are required by default: a load failure aborts startup unless the module says `optional=True`), its `routes`, an optional `health_check`, its `migrations_path` and `depends_on` ordering, and the `capabilities` it needs. Reference other modules' entities by plain UUID — never declare cross-module foreign keys. #### `ModuleDescriptor.effective_capabilities` ``` def effective_capabilities(self) -> list[CapabilityRequirement] ``` Capabilities to preflight at load (the declared list). #### `ModuleDescriptor.migration_chain_key` ``` def migration_chain_key(self) -> str ``` Key of this module's migration chain (version-table suffix). #### `ModuleDescriptor.database_instances` ``` def database_instances(self) -> set[str] ``` The database instance this module's tables live in (one element). Resolution: 1. `migrations_database` when set (explicit, always wins); 2. otherwise the instances of the REQUIRED `database` requirements: none -\> `default`; exactly one -\> that one; several -\> `default` if it is among them (a module that reads `default` and also uses `analytics` keeps its tables in `default`), else ambiguous -\> `MigrationTargetError` (declare `migrations_database`). Optional (`required=False`) database requirements never host the chain, so an optional, unconfigured instance cannot break `upgrade-all`. A module's chain is applied to exactly ONE database — applying it to every declared instance duplicated its tables. ### `ModuleLoadSummary` ``` class ModuleLoadSummary( loaded: list[ModuleDescriptor] = , skipped: list[SkippedModule] = , failed: list[FailedModule] = , duplicates: list[str] = , explicit: list[str] = , consumes: dict[str, list[dict[str, Any]]] = , manifests: dict[str, Any] = , ) -> None ``` Aggregate result of a load pass: loaded, skipped, failed, and duplicates. #### `ModuleLoadSummary.success` ``` def success(self) -> bool ``` #### `ModuleLoadSummary.to_dict` ``` def to_dict(self) -> dict[str, Any] ``` ### `SkippedModule` ``` class SkippedModule(name: str, reason: str) -> None ``` A module the loader skipped, with the reason why. ## Examples and related guides - [Add a module]() - [Python API overview]()