Add a contract¶

Goal: module A answers a synchronous question or runs a command for module B, through a versioned contract such as stats.quota.v1, with a typed stub that B imports instead of A's code.

Use it when the caller needs an answer now: a quota check, a lookup, a command that must succeed or fail before the caller goes on.

Do not use it when the caller only needs to tell others that something happened (publish an event), or to read many rows of another module for a list page (build a projection). A contract call runs in the provider's own unit of work, never in the caller's transaction: a command is not atomic with the caller's write. A multi-step change with undo is a saga.

Prerequisites¶

The providing module exists (Add a module).
If contracts.py already exists, it must parse and import command, contract from sandloom.sdk.

Steps¶

1. Run the generator in the providing module¶

sandloom add contract stats quota

The contract name is <module>.<name>.v1, here stats.quota.v1. The generator writes, inside the module package:

the provider class, appended to <module>/contracts.py;
its DTO, a pydantic model appended to <module>/schemas.py: a provider authors the models and Enums its signatures use in its own schemas.

Then sandloom add runs sandloom generate, which writes the generated files in the shared DTO package:

the typed stub (<app>/shared/contracts/stats_quota_v1.py), rendered from the provider;
a wire-shape copy of the module's contract DTOs (<app>/shared/contracts/stats_dto.py), which the stub names and callers receive.

Generated files are never edited by hand; sandloom generate rewrites them.

2. Read what it wrote¶

The provider, with two stand-in methods. A public async def is a read; @command(idempotency_key=...) marks a command, deduplicated by that argument and committed in the provider's own unit:

"""Contracts the stats module provides."""

from sandloom.sdk import command, contract

from .schemas import QuotaStatus


@contract("stats.quota.v1")
class Quota:
    """Contract ``stats.quota.v1``: public ``async def`` methods are reads,
    ``@command`` marks a command (deduplicated by its key, a saga step)."""

    async def status(self) -> QuotaStatus:
        return QuotaStatus(ready=True)

    @command(idempotency_key="operation_key")
    async def record(self, amount: int, operation_key: str) -> int:
        return amount

The stub callers import: a Protocol and a typed reference to it, which names the DTO's generated copy:

# Generated by `sandloom generate` (Public API 1.0): never edit by hand.
"""Contract ``stats.quota.v1``: the typed stub callers import.

Generated from ``demo.modules.stats.contracts.Quota``.
"""

from __future__ import annotations

from typing import Protocol

from demo.shared.contracts.stats_dto import QuotaStatus
from sandloom.core.descriptors.contract_refs import ContractRef


class StatsQuotaV1(Protocol):
    """Contract ``stats.quota.v1``: public ``async def`` methods are reads,
    ``@command`` marks a command (deduplicated by its key, a saga step).
    """

    async def record(self, amount: int, operation_key: str) -> int: ...
    async def status(self) -> QuotaStatus: ...


STATS_QUOTA_V1: ContractRef[StatsQuotaV1] = ContractRef("stats.quota.v1", StatsQuotaV1)

The copy carries the fields, their types and constraints, defaults, aliases, Literal / Enum values, nested models and the model_config options; not methods or validators (they stay the provider's):

# Generated by `sandloom generate` (Public API 1.0): never edit by hand.
"""Contract DTOs of module ``stats``: the wire-shape copies callers import.

Generated from the provider's own classes in:

- ``demo.modules.stats.schemas``

A copy carries the fields, their types and constraints, defaults,
aliases, Literal / Enum values, nested models and model_config options;
not the methods, properties or validators (the provider's own).
"""

from __future__ import annotations

from pydantic import BaseModel

from sandloom.core.descriptors.contract_refs import dto_copy


@dto_copy("demo.modules.stats.schemas:QuotaStatus")
class QuotaStatus(BaseModel):
    """What ``stats.quota.v1.status`` returns. A contract DTO lives in its
    provider; `sandloom generate` copies it into the shared DTO package.
    """

    ready: bool

3. Write the real methods¶

Replace the stand-ins with your API. The class stays stateless: no __init__, no instance attributes (ClassVar constants are allowed). Every parameter and return type must be a wire type; models and Enums go in your module's schemas, inside the closed set of wire shapes (Contracts). A caller receives the generated copy, locally and over RPC; the provider receives its own class.

The tutorial app's stats module provides one read:

@contract("stats.quota.v1")
class Quota:
    """Contract ``stats.quota.v1``: public ``async def`` methods are reads,
    ``@command`` marks a command (deduplicated by its key, a saga step)."""

    async def may_create_item(self) -> bool:
        return await services.item_count() < services.ITEM_LIMIT

4. Regenerate the stub¶

sandloom generate regenerates the stub and the DTO copies from the provider. It compares the new stub with the file on disk first: an added method is written, but a removed or changed method is refused (SANDLOOM-T007), because callers rely on it, and nothing is written. sandloom test never writes: it fails on a stale file ("run sandloom generate").

Replacing the generator's stand-ins removes status and record. While no caller uses the contract yet and the stub is not committed, delete the stub file and generate it again:

rm <app>/shared/contracts/stats_quota_v1.py
sandloom generate stats

The result in the tutorial app:

# Generated by `sandloom generate` (Public API 1.0): never edit by hand.
"""Contract ``stats.quota.v1``: the typed stub callers import.

Generated from ``tasklist.modules.stats.contracts.Quota``.
"""

from __future__ import annotations

from typing import Protocol

from sandloom.core.descriptors.contract_refs import ContractRef


class StatsQuotaV1(Protocol):
    """Contract ``stats.quota.v1``: public ``async def`` methods are reads,
    ``@command`` marks a command (deduplicated by its key, a saga step).
    """

    async def may_create_item(self) -> bool: ...


STATS_QUOTA_V1: ContractRef[StatsQuotaV1] = ContractRef("stats.quota.v1", StatsQuotaV1)

Once the contract is committed and called, change it only compatibly (add methods); a breaking change is a new contract, stats.quota.v2.

5. Call it from another module¶

Declare the caller once, at module level, with the generated reference. define_module adds it to the module's consumes, and startup fails if the contract is neither provided in the process nor bound to a remote service (SANDLOOM-B005):

# Required: items must never be created without the quota check.
quota = Contract(STATS_QUOTA_V1)

Then await its methods. The call is typed by the Protocol:

@router.post("/items", status_code=201)
async def create_item(body: ItemIn, user: CurrentUser) -> ItemOut:
    if not await quota.may_create_item():
        raise HTTPException(status_code=402, detail="item quota reached")
    return await services.create_item(body, owner_id=user.user_id)


@router.get("/items")
async def list_items(
    user: CurrentUser, limit: Annotated[int, Query(ge=1, le=100)] = 20
) -> list[ItemOut]:
    """A page of the tenant's items; the limit caps each request."""
    return await services.list_items(limit=limit)


@router.get("/items/{item_id}")
async def get_item(item_id: UUID, user: CurrentUser) -> ItemOut:
    item = await services.get_item(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="item not found")
    return item

The call works the same whether the provider runs in the same process or in another service behind RPC (configured with SERVICE_BINDINGS; see Service RPC). A provider that is deployed but down answers 503.

6. Required or optional¶

Contract(REF) is required. Use it whenever the caller has no correct answer without the provider. A gate whose absence must deny, such as the quota above, is always required: an absent quota service must not silently allow writes.

Contract(REF, optional=True) is for the rare caller that has a correct business answer without the provider, such as a feature that is off or an enrichment that is skipped:

recommendations = Contract(RECO_ITEMS_V1, optional=True)

client = recommendations.client()   # RecoItemsV1 | None
extra = [] if client is None else await client.for_item(item_id)

client() returns None only when the provider is neither local nor bound in this deployment, which is decided at startup. It does not hide an outage: a deployed provider that fails still raises.

Verify¶

Test the caller with a stand-in for the contract, so the test does not need the provider:

async def test_create_and_read_an_item(client: Any) -> None:
    _, headers = as_user(tenant_id=uuid4())
    with fake_contract(STATS_QUOTA_V1, Quota(allow=True)):
        created = await client.post(
            "/api/v1/tasks/items", json={"title": "first"}, headers=headers
        )
    assert created.status_code == 201, created.text
    assert len(published(ItemCreated)) == 1
    item = created.json()
    fetched = await client.get(f"/api/v1/tasks/items/{item['id']}", headers=headers)
    assert fetched.status_code == 200
    assert fetched.json()["title"] == "first"


async def test_the_quota_refuses_a_new_item(client: Any) -> None:
    _, headers = as_user(tenant_id=uuid4())
    with fake_contract(STATS_QUOTA_V1, Quota(allow=False)):
        response = await client.post(
            "/api/v1/tasks/items", json={"title": "x"}, headers=headers
        )
    assert response.status_code == 402
    assert published(ItemCreated) == []


async def test_items_need_an_authenticated_user(client: Any) -> None:
    response = await client.post("/api/v1/tasks/items", json={"title": "x"})
    assert response.status_code == 401

Then run both modules' tests and the gates:

sandloom test stats
sandloom test tasks

sandloom generate also writes contracts.snapshot.json for the provider; commit it with the stub and the DTO copy.

Errors you may hit¶

SANDLOOM-B005: a required contract is neither provided in this process nor bound in SERVICE_BINDINGS.
SANDLOOM-B018: the provider class has state, a non-async member, a @command without idempotency_key, or a base class outside the module.
SANDLOOM-C012: a signature DTO is outside the closed set of wire shapes (the field path and the fix are named), belongs to another module or a function, or its generated copy is missing or hand-edited (run sandloom generate).
SANDLOOM-R005: the contract is not available in this deployment at call time.
SANDLOOM-T004: the provider's contract snapshot changed incompatibly.
SANDLOOM-T007: the stub is out of date, or would lose or change a method.

Contracts: reads, commands, wire types, local equals remote.
Service RPC: calling a provider in another service.
Add a saga: command steps with compensation.
Testing modules.