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

Start with (beginner)¶

Advanced¶

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] = <factory>,
    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 <owner>.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 <module>.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 <module>:<name> (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 = "<module>.<name>") 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 <module>: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] = <factory>,
    capabilities: list[CapabilityRequirement] = <factory>,
    migrations_database: str | None = None,
    migrations_chain: str | None = None,
    consumes: list[ContractRequirement] = <factory>,
    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] = <factory>,
    skipped: list[SkippedModule] = <factory>,
    failed: list[FailedModule] = <factory>,
    duplicates: list[str] = <factory>,
    explicit: list[str] = <factory>,
    consumes: dict[str, list[dict[str, Any]]] = <factory>,
    manifests: dict[str, Any] = <factory>,
) -> 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