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 <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¶

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.<method>( 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