# `sandloom.sdk.contracts` Sync contracts: typed providers and callers, `contract_method` and its helpers, and (advanced) the provider runtime. Public API 1.0 (spec §2.2, Appendix C). `from sandloom.sdk.contracts 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 | | --- | --- | --- | | [`command`](<#command>) | function | Mark a provider method as a command: deduplicated by its `idempotency_key` argument, committed in the provider's own unit, not atomic wit... | | [`Contract`](<#contract>) | function | The caller of contract `ref` (spec §5.3), declared at module level: | | [`contract`](<#contract_1>) | function | Declare a stateless provider class of contract `name`: | ## Advanced | Name | Kind | Summary | | --- | --- | --- | | [`BACKWARD`](<#backward>) | constant | | | [`bind_contract`](<#bind_contract>) | function | Wrap `impl` in a `ContractProvider` (idempotent). | | [`caller_user_id`](<#caller_user_id>) | function | The end user on whose behalf the running contract call executes, or `None` (a system / service caller, a task, a consumer). | | [`contract_dto_problems`](<#contract_dto_problems>) | function | DTOs of contract `name` no caller process could hold (SANDLOOM-C012). | | [`contract_error`](<#contract_error>) | function | Build the caller-side exception for a contract error (local or RPC). | | [`contract_kind`](<#contract_kind>) | function | How a BOUND contract member runs: `read` / `command`, or `None` for a plain (non-contract) member of the provider. | | [`contract_method`](<#contract_method>) | function | Mark an async method as a service-contract entry point. | | [`contract_read_scope`](<#contract_read_scope>) | function | Open a `ContractReadScope` for the block (reuses an outer one). | | [`contract_reads`](<#contract_reads>) | function | The declared-operations handle (`sandloom.core.orm.Reads`) of the running contract call. | | [`contract_schema`](<#contract_schema>) | function | The validated `ContractSchema` of a contract provider (cached). | | [`contract_session_bound`](<#contract_session_bound>) | function | Whether a test seam session (`bind_contract_session`) is bound. | | [`contract_spec`](<#contract_spec>) | function | The `ContractMethodSpec` of a (bound) contract method, if any. | | [`ContractMethodRuntime`](<#contractmethodruntime>) | class | The validated, wire-portable shape of one exposed contract method. | | [`ContractMethodSpec`](<#contractmethodspec>) | class | How the contract runtime runs one contract method. | | [`ContractProvider`](<#contractprovider>) | class | A registered contract bound to its provider's core and database. | | [`ContractReadScope`](<#contractreadscope>) | class | A request-scoped scope in which LOCAL read-contract calls share work. | | [`ContractReadScopeMiddleware`](<#contractreadscopemiddleware>) | class | Pure-ASGI middleware opening a `ContractReadScope` per request. | | [`ContractRef`](<#contractref>) | class | A typed reference to a contract (emitted by the stub generator): \`STATS\_QUOTA\_V1: ContractRef\[StatsQuotaV1\] = ContractRef("stats.quota.v1... | | [`ContractSchema`](<#contractschema>) | class | Every exposed method of one registered contract provider + a stable hash. | | [`current_contract_read_scope`](<#current_contract_read_scope>) | function | The read scope bound to the current context, if any. | | [`DEFAULT_CONTRACT_DTO_PACKAGES`](<#default_contract_dto_packages>) | constant | | | [`FORWARD`](<#forward>) | constant | | | [`invalidate_contract_read_memo`](<#invalidate_contract_read_memo>) | function | Clear the current read scope's memo (a write happened elsewhere). | | [`normalize_contract_error`](<#normalize_contract_error>) | function | The caller-side form of a provider exception, or `None` (a 5xx bug). | | [`OptionalContract`](<#optionalcontract>) | class | `Contract(REF, optional=True)`: `client()` is `None` only when the provider is neither local nor bound in this deployment (decided at sta... | | [`run_contract_call`](<#run_contract_call>) | async function | Run one contract call in the provider's own Unit of Work. | | [`service`](<#service>) | function | The proxy of contract `name` (spec §2.3, §5.3). | ## Beginner names ### `command` ``` def command( *, idempotency_key: str, idempotency_scope: tuple[str, ...] = (), database: str | None = None, risk: Literal['money'] | None = None, ) -> Callable[[Callable[PS, R]], Callable[PS, R]] ``` Mark a provider method as a command: deduplicated by its `idempotency_key` argument, committed in the provider's own unit, not atomic with the caller (a saga step): ``` @command(idempotency_key="operation_key") async def reserve(self, amount: int, operation_key: str) -> int: ... ``` `risk="money"` declares a money source (ADR-0009). ### `Contract` ``` def Contract(ref: ContractRef[P], *, optional: bool = False) -> P | OptionalContract[P] ``` The caller of contract `ref` (spec §5.3), declared at module level: ``` quota = Contract(STATS_QUOTA_V1) # required: startup fails without it if not await quota.may_create_note(): ... ``` `define_module` adds it to the module's `consumes` (name and `required` exact; `methods` from the direct `quota.(` calls in the defining module, else every method). ### `contract` ``` def contract(name: str, *, database: str | None = None) -> Callable[[C], C] ``` Declare a stateless provider class of contract `name`: ``` @contract("stats.quota.v1") class Quota: async def may_create_note(self) -> bool: ... ``` Public `async def` methods are reads; `@command(...)` marks commands; `@contract.read(planned=..., database=...)` keeps today's read options. ## Advanced names ### `BACKWARD` ``` BACKWARD: str = 'undo' ``` *(no docstring)* ### `bind_contract` ``` def bind_contract(impl: Any, *, core: Any, database: str, name: str) -> Any ``` Wrap `impl` in a `ContractProvider` (idempotent). ### `caller_user_id` ``` def caller_user_id() -> UUID | None ``` The end user on whose behalf the running contract call executes, or `None` (a system / service caller, a task, a consumer). Locally the authenticated subject of the calling request; over RPC the same subject, carried hop after hop in the signed caller facts of each call's context (the provider binds them as its call's request scope). A provider holds a personal resource to it (billing's personal spend, Public API 1.0 spec §7.4). ### `contract_dto_problems` ``` def contract_dto_problems( name: str, impl: Any, *, provider_module: str | None, module_packages: Iterable[str] = (), shared_packages: Iterable[str] = (), ) -> list[str] ``` DTOs of contract `name` no caller process could hold (SANDLOOM-C012). A provider authors its contract DTOs (pydantic models, Enums) in its own module package; `sandloom generate` copies each one a signature names into the shared DTO package, where every caller imports the copy (the stub names it, and a remote caller decodes into it). A DTO a module package holds is therefore allowed in its OWN provider's signatures; what no copy can come from is refused: a class of ANOTHER module's package (use its generated copy from the shared DTO package), or one defined in a function or `__main__`. Classes of a shared package (`sandloom.shared`, `CONTRACT_DTO_PACKAGES`) are referenced as they are. The closed set of DTO shapes is checked when `@contract` decorates the provider. Checked only for contracts a MODULE registers (`provider_module` known); kernel-level test registrations are exempt. ### `contract_error` ``` def contract_error( *, status_code: int, detail: Any, headers: Mapping[str, str] | None = None, code: str | None = None, ) -> ContractCallError ``` Build the caller-side exception for a contract error (local or RPC). ### `contract_kind` ``` def contract_kind(method: Any) -> Literal['read', 'command'] | None ``` How a BOUND contract member runs: `read` / `command`, or `None` for a plain (non-contract) member of the provider. ### `contract_method` ``` def contract_method( fn: _F | None = None, *, command: bool = False, idempotency_key: str | None = None, idempotency_scope: tuple[str, ...] = (), database: str | None = None, planned: bool = False, risk: Literal['admin', 'money', 'security'] | None = None, ) -> _F | Callable[[_F], _F] ``` Mark an async method as a service-contract entry point. Every call — local (`core.get_service`) or remote (internal RPC server) — runs through `run_contract_call`: a command's `idempotency_key` argument is required, and the method body runs in the provider's own Unit of Work (`contract_session`). Stack it UNDER `@classmethod`. `command=True` commits the provider's UoW; such a call is NOT atomic with the caller's transaction (saga semantics — the caller compensates or uses events). `database` only applies to direct calls of an UNREGISTERED provider (tests): a direct command call of a registered provider runs on its registration's database under its contract name, sharing the registry path's dedupe record. `idempotency_scope` partitions a command's operation keys by named domain arguments, for example `("tenant_type", "tenant_id")` for personal and organization billing accounts. Changing this tuple changes the dedupe namespace; retain domain UNIQUE operation keys when rolling out a change. The signature must be wire-portable (checked when the provider is registered): every parameter and the return annotated with a contract wire type (`sandloom.core.contract_wire`), no `*args` / `**kwargs`. `planned=True` (reads only) declares a contract whose body runs only declared planner operations through `contract_reads`: no Unit of Work is opened; the reads lease from the contract pool in a read-only scope (`publish_event` / `record_audit` raise), with the tenant bound exactly as for any contract call (the caller's locally, the verified RPC context remotely). `risk` (commands only) declares the risk of the command's effect at its source, once (ADR-0009 item 7): `risk="money"` for a command that moves money. It is recorded in the contract's describe document and snapshot; a local call in a request records the effect there (its own `idempotency_key` is its dedupe record), which triggers the automatic audit row of a sensitive write. ### `contract_read_scope` ``` def contract_read_scope( *, memo: bool | None = None, idle_ms: int | None = None, ) -> AsyncIterator[ContractReadScope] ``` Open a `ContractReadScope` for the block (reuses an outer one). The framework opens one per HTTP request; open one explicitly in a worker / script that makes several local read-contract calls in a row. Memoisation is on unless `memo=False`; `idle_ms` defaults to `CONTRACT_READ_SCOPE_IDLE_MS`, and `idle_ms < 0` holds the shared connection until the block ends. ### `contract_reads` ``` def contract_reads() -> Any ``` The declared-operations handle (`sandloom.core.orm.Reads`) of the running contract call. In a PLANNED read contract (`@contract_method(planned=True)`) the reads run in the call's read-only scope on the provider's contract pool (one shared read-only transaction for general-executor reads; the optimized executor when enabled and eligible) — never in the caller's transaction, locally and over RPC alike. In any other contract call they run on `contract_session`. Under the test seam (`bind_contract_session`) they run on the bound test session. ### `contract_schema` ``` def contract_schema(impl: Any) -> ContractSchema ``` The validated `ContractSchema` of a contract provider (cached). Every public member must be a `contract_method` with a wire-portable signature, or a `ClassVar` constant (spec §5.3: never exposed); private members (`_name`) are never exposed. Raises `sandloom.core.contract_wire.ContractTypeError` listing every problem — called at registration, so a non-portable contract never deploys. ### `contract_session_bound` ``` def contract_session_bound() -> bool ``` Whether a test seam session (`bind_contract_session`) is bound. Contract calls then all run on that ONE session, so callers must not run them concurrently. ### `contract_spec` ``` def contract_spec(method: Any) -> ContractMethodSpec | None ``` The `ContractMethodSpec` of a (bound) contract method, if any. ### `ContractMethodRuntime` ``` class ContractMethodRuntime( name: str, kind: Literal['read', 'command'], signature: inspect.Signature, annotations: dict[str, Any], returns: Any, idempotency_key: str | None = None, idempotency_scope: tuple[str, ...] = (), risk: str | None = None, ) -> None ``` The validated, wire-portable shape of one exposed contract method. Built once at registration; the local path and the RPC server bind, validate and encode every call through it, so both see the same argument rules and the same result types. #### `ContractMethodRuntime.describe` ``` def describe(self) -> dict[str, Any] ``` JSON-able schema published by the RPC describe endpoint. #### `ContractMethodRuntime.bind` ``` def bind( self, label: str, args: Sequence[Any], kwargs: Mapping[str, Any], *, local: bool = False, ) -> tuple[tuple[Any, ...], dict[str, Any]] ``` Bind + validate UNTAGGED arguments exactly like a Python call would. Returns `(positional_only_args, keyword_args)`. Unknown / missing arguments, positional-only or keyword-only violations and values that do not validate against their annotation raise `ContractArgumentError` (422) before the provider runs. `local=True` binds the caller's ORIGINAL values of an in-process call: each goes through its parameter's `WirePlan` (immutable exact-type values pass through, fully typed ones round-trip through pydantic-core, the rest through the tagged tree) — what the provider would receive over RPC. #### `ContractMethodRuntime.decode_result` ``` def decode_result(self, wire: Any, codec: int = 1) -> Any ``` Rebuild a result from its wire form using the return annotation. #### `ContractMethodRuntime.local_result` ``` def local_result(self, label: str, result: Any) -> Any ``` What an in-process caller receives for `result` (see bind(local=)). ### `ContractMethodSpec` ``` class ContractMethodSpec( kind: Literal['read', 'command'] = 'read', idempotency_key: str | None = None, database: str | None = None, planned: bool = False, idempotency_scope: tuple[str, ...] = (), risk: str | None = None, ) -> None ``` How the contract runtime runs one contract method. #### `ContractMethodSpec.readonly` ``` readonly (property) ``` ### `ContractProvider` ``` class ContractProvider(impl: Any, *, core: Any, database: str, name: str) -> None ``` A registered contract bound to its provider's core and database. Returned by `core.get_service` for a locally provided contract and used by the internal RPC server, so both paths run every call through `run_contract_call`. Only the contract's exposed methods are visible (exactly what a remote proxy can call); each local call is bound with Python's rules, its arguments and result round-trip through the wire codec and are validated against the annotations, and provider errors surface as `ContractCallError` — so a local call is observably identical to a remote one. #### `ContractProvider.contract_impl` ``` contract_impl (property) ``` The registered provider object this binding wraps. #### `ContractProvider.contract_schema` ``` contract_schema (property) ``` The exposed methods (and hash) of this contract. #### `ContractProvider.contract_name` ``` contract_name (property) ``` #### `ContractProvider.call_local` ``` async def call_local( self, runtime: ContractMethodRuntime, args: tuple[Any, ...], kwargs: dict[str, Any], ) -> Any ``` An in-process call: exactly what the provider/caller see over RPC. Arguments and result go through their :class:`~sandloom.core.contract_wire. WirePlan` (validated copies; immutable exact-type values pass through untouched). Inside a `ContractReadScope` an identical read call is served from the scope's memo, and any command clears it. #### `ContractProvider.encode_result` ``` def encode_result( self, runtime: ContractMethodRuntime, result: Any, *, typed: bool = False, ) -> Any ``` The tagged wire form of `result` (see `encode_result_as`). #### `ContractProvider.encode_result_as` ``` def encode_result_as( self, runtime: ContractMethodRuntime, result: Any, *, typed: bool = False, ) -> tuple[int, Any] ``` `(codec, wire)` of `result`; codec v2 when `typed` and it fits. #### `ContractProvider.invoke` ``` async def invoke( self, runtime: ContractMethodRuntime, args: Sequence[Any], kwargs: Mapping[str, Any], ) -> Any ``` Bind + validate UNTAGGED arguments and run the provider (both paths). Raises `ContractCallError` for argument errors and for every provider `HTTPException` / `http_status` exception; anything else is a provider failure and propagates unchanged. ### `ContractReadScope` ``` class ContractReadScope(*, memo: bool = True, idle_seconds: float | None = 0.0) -> None ``` A request-scoped scope in which LOCAL read-contract calls share work. Opened per HTTP request by the framework (`ContractReadScopeMiddleware`) or explicitly with `contract_read_scope`. Inside it: - all local read-contract calls on the same database reuse ONE connection / transaction from the provider's contract pool (READ COMMITTED: every statement sees the latest committed data, exactly like a fresh UoW per call). It is NEVER the caller's transaction, so — as over RPC — the provider does not see the caller's uncommitted writes. It is always rolled back: when the scope ends, when a call raises or leaves ORM writes pending, and once the request stops making read calls — `idle_seconds=0` (default) releases it on the next event-loop turn after a call, i.e. as soon as the request awaits anything else, so back-to-back calls share it but it never idles in the pool's absence while the request does other I/O; `>0` keeps it that long; `None` holds it until the scope ends. A call that finds the shared connection busy (concurrent tasks, a read contract calling another on the same database) falls back to its own Unit of Work. A call that finds the shared transaction aborted by an earlier call's swallowed database error (SQLSTATE 25P02) drops that connection and reruns in its own Unit of Work (`stats["replaced"]`); - identical read calls (local or RPC; same contract, method, bound tenant and arguments) are memoised for the rest of the scope — each caller still receives its own copy. Any command call (local or remote) clears the memo. External commits are visible after invalidation or in a new scope; use `memo=False` when every read must be fresh. A read contract must not write: a raw DML statement it executed would be visible to later read contracts of the same scope until rollback. #### `ContractReadScope.stats` ``` stats (property) ``` Counters (diagnostics / performance budget tests). #### `ContractReadScope.memo_enabled` ``` memo_enabled (property) ``` #### `ContractReadScope.memo_get` ``` def memo_get(self, key: Any) -> Any ``` #### `ContractReadScope.memo_set` ``` def memo_set(self, key: Any, value: Any) -> None ``` #### `ContractReadScope.invalidate` ``` def invalidate(self) -> None ``` Forget memoised reads (a command may have changed what they return). #### `ContractReadScope.run` ``` async def run( self, func: Callable[..., Awaitable[Any]], args: tuple[Any, ...], kwargs: dict[str, Any], *, core: Any, database: str, qualname: str, ) -> Any ``` Run a read contract on the shared unit, or `_SCOPE_FALLBACK`. #### `ContractReadScope.release_connections` ``` async def release_connections(self) -> None ``` Roll back and release every shared connection NOW; the scope stays usable (a later read reopens one). The operation-intent windows (`external_call` / `release_check`) release it first (spec §4.6). A read still running on a shared connection (another task) is waited for, then its unit is closed: no read connection survives the drain. A read that cannot finish first — the caller runs inside it — is refused (SANDLOOM-R008) instead of waited for. #### `ContractReadScope.aclose` ``` async def aclose(self) -> None ``` End the scope: roll back and release every shared connection. ### `ContractReadScopeMiddleware` ``` class ContractReadScopeMiddleware(app: Any) -> None ``` Pure-ASGI middleware opening a `ContractReadScope` per request. Lazy: nothing is checked out unless the request makes a local read contract call. Disabled by `CONTRACT_READ_SCOPE_ENABLED=false`. ### `ContractRef` ``` class ContractRef(Generic)(name: str, protocol: type[P] | None = None) -> None ``` A typed reference to a contract (emitted by the stub generator): `STATS_QUOTA_V1: ContractRef[StatsQuotaV1] = ContractRef("stats.quota.v1", StatsQuotaV1)`. `protocol` is the stub's Protocol: a `Contract(REF)` caller receives results as its methods' annotations name them (the generated DTO copies), locally and over RPC. `None` (a hand-written reference): results as the runtime returns them. ### `ContractSchema` ``` class ContractSchema(methods: dict[str, ContractMethodRuntime], hash: str) -> None ``` Every exposed method of one registered contract provider + a stable hash. #### `ContractSchema.describe` ``` def describe(self) -> dict[str, Any] ``` ### `current_contract_read_scope` ``` def current_contract_read_scope() -> ContractReadScope | None ``` The read scope bound to the current context, if any. ### `DEFAULT_CONTRACT_DTO_PACKAGES` ``` DEFAULT_CONTRACT_DTO_PACKAGES: tuple = ('sandloom.shared',) ``` *(no docstring)* ### `FORWARD` ``` FORWARD: str = 'do' ``` *(no docstring)* ### `invalidate_contract_read_memo` ``` def invalidate_contract_read_memo() -> None ``` Clear the current read scope's memo (a write happened elsewhere). ### `normalize_contract_error` ``` def normalize_contract_error(exc: BaseException) -> ContractCallError | None ``` The caller-side form of a provider exception, or `None` (a 5xx bug). `HTTPException` and `http_status`-protocol exceptions become a `ContractCallError` with the same status, JSON-encoded detail, headers and code — exactly what the RPC server transmits. ### `OptionalContract` ``` class OptionalContract(Generic)(ref: ContractRef[P]) -> None ``` `Contract(REF, optional=True)`: `client()` is `None` only when the provider is neither local nor bound in this deployment (decided at startup); a deployed provider that is down still raises (503). Use it only where the caller has a correct answer without the provider (a feature off, an enrichment skipped); a gate whose absence must deny is a required `Contract(REF)` (startup fails without it, SANDLOOM-B005). #### `OptionalContract.contract_name` ``` contract_name (property) ``` #### `OptionalContract.required` ``` required (property) ``` #### `OptionalContract.client` ``` def client(self) -> P | None ``` The provider's proxy, or `None` when it is not deployed. Its results arrive as the stub annotates them, like `Contract(REF)`'s. ### `run_contract_call` ``` async def run_contract_call( func: Callable[..., Awaitable[Any]], args: tuple[Any, ...], kwargs: dict[str, Any], *, spec: ContractMethodSpec, core: Any = None, database: str = 'default', contract: str | None = None, ) -> Any ``` Run one contract call in the provider's own Unit of Work. The single code path for local and RPC-served contract calls: enforces a command's idempotency key and deduplicates commands through the provider database's durable command log (claimed and completed inside the command's own UoW), opens the provider's UoW on `database` (rolled back for a read, committed + outbox dispatched for a command) and exposes it via `contract_session`. `contract` is the registered contract name (the dedupe namespace; defaults to the method's qualified name). ### `service` ``` def service(name: str, *, optional: bool = False) -> Any ``` The proxy of contract `name` (spec §2.3, §5.3). `optional=True` returns `None` only when the provider is neither local nor bound in this deployment; a deployed provider that is down still raises when called. ## Examples and related guides - [Provide and call a contract]() - [Python API overview]()