# `sandloom.sdk.web` Routes: sessions and dependencies, the caller types, route deviations and the operation-intent primitives. Public API 1.0 (spec §2.2, Appendix C). A name whose mechanism lands at a later hand-off raises `NotImplementedError` naming it. `from sandloom.sdk.web 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 | | --- | --- | --- | | [`CurrentUser`](<#currentuser>) | annotated type | | | [`module_router`](<#module_router>) | function | The module's router (spec §5.1), in `/api.py`: | | [`public`](<#public>) | function | Declare a route unauthenticated, and why (ADR-0009 item 1). | ## Advanced | Name | Kind | Summary | | --- | --- | --- | | [`allow_multi_db_write`](<#allow_multi_db_write>) | function | The reviewed best-effort 1PC opt-out of the one-writer rule (spec §4.4 rule 4): the listed secondaries may write too; they commit first a... | | [`at_commit`](<#at_commit>) | function | Re-check the caller's session before the COMMIT of EVERY unit the request opens, empty ones included (spec §4.5; today's \`revocation\_at\_c... | | [`authenticates`](<#authenticates>) | function | Mark `fn` (a FastAPI dependency) as authenticating the caller. | | [`authorizes`](<#authorizes>) | function | Mark `fn` (a FastAPI dependency) as a custom authorization check. | | [`Capability`](<#capability>) | function | Annotated dependency resolving any capability instance. | | [`current_service`](<#current_service>) | async function | The service principal of an internal call (`SOURCE_INTERNAL`): the internal token authenticates it (404 otherwise, existence hidden). A r... | | [`current_user_id`](<#current_user_id>) | async function | Extract the authenticated user id from the Bearer access token. | | [`CurrentService`](<#currentservice>) | annotated type | | | [`Database`](<#database>) | function | Annotated dependency for a named database session. | | [`DbSession`](<#dbsession>) | annotated type | | | [`external_call`](<#external_call>) | function | Run one external (PSP) call with no unit of the request open (§4.6). | | [`get_client_ip`](<#get_client_ip>) | function | Return the real client IP for `request` (Starlette `Request`-like). | | [`idempotency_guard`](<#idempotency_guard>) | function | Public re-export of the Idempotency-Key dependency factory. | | [`ModuleRouter`](<#modulerouter>) | class | An `APIRouter` whose route decorators also take `tenant: Literal["required", "optional", "none"] \\| None = None`. | | [`no_idempotency`](<#no_idempotency>) | function | Opt a write route out of the automatic idempotency guard (streaming, non-JSON bodies). The reason is recorded in the route snapshot. | | [`operation_intent`](<#operation_intent>) | function | Declare the operation-intent mode (spec §4.6): no request unit, sequential `open_uow()` units, external calls in `external_call()`, and \`... | | [`rate_limiter`](<#rate_limiter>) | function | Expose rate limiter dependency routed through sandloom.core.ingress.rate\_limit. | | [`ReadOnlyDbSession`](<#readonlydbsession>) | annotated type | | | [`release_check`](<#release_check>) | async function | The final re-read before an intent route releases a bearer value. | | [`replay_safe`](<#replay_safe>) | function | Opt a keyed write in to BODY replay (idempotency mode B, spec §4.7). | | [`request_operation_key`](<#request_operation_key>) | function | The request's operation key: a stable digest of its route, caller, tenant and `Idempotency-Key` as claimed by the route's idempotency gua... | | [`require_platform_admin`](<#require_platform_admin>) | async function | | | [`ResolvedAccount`](<#resolvedaccount>) | class | The billing account a release check is for (spec §2.3, §4.6, §7). | | [`sensitive`](<#sensitive>) | function | The escape hatch for a route whose authority check is hand-rolled: declare its risk (`admin` / `money` / `security`) so the framework app... | | [`TenantId`](<#tenantid>) | annotated type | | | [`TenantScope`](<#tenantscope>) | function | FastAPI dependency that binds the request's tenant for its whole scope. | | [`writes`](<#writes>) | function | Name the route's write store when it injects several writable sessions (spec §6.1, SANDLOOM-B009). | ## Beginner names ### `CurrentUser` ``` CurrentUser = Annotated[sandloom.shared.subject.UserContext, Depends(dependency=, use_cache=True, scope=None)] ``` *(no docstring)* ### `module_router` ``` def module_router( *, prefix: str | None = None, tags: Sequence[str] | None = None, tenant: TenantBinding = 'derive', dependencies: Sequence[params.Depends] = (), ) -> ModuleRouter ``` The module's router (spec §5.1), in `/api.py`: ``` router = module_router() @router.post("/", status_code=201) async def create_note(body: NoteIn, user: CurrentUser) -> NoteOut: ... ``` `prefix` defaults to `/api/v1/` and `tags` to `[]`. `tenant="derive"` binds the tenant (`required`) when the module declares a `TenantScoped` model, else none; a route's own `tenant=` overrides. The binding scopes data only; it never selects a billing account. ### `public` ``` def public(*, reason: str, issues_tokens: bool = False) -> params.Depends ``` Declare a route unauthenticated, and why (ADR-0009 item 1). `dependencies=[public(reason="liveness probe")]`. A token issuer (login, refresh, SSO callback) adds `issues_tokens=True`: its live issuance check is unchanged, and an explicit idempotency guard on it must be the release-and-re-execute form (SANDLOOM-B019). A PSP webhook is public too; its verified signature, checked in the endpoint, is its authorization. ## Advanced names ### `allow_multi_db_write` ``` def allow_multi_db_write(*, secondaries: tuple[str, ...], reason: str) -> params.Depends ``` The reviewed best-effort 1PC opt-out of the one-writer rule (spec §4.4 rule 4): the listed secondaries may write too; they commit first and the write store last. ### `at_commit` ``` def at_commit() -> params.Depends ``` Re-check the caller's session before the COMMIT of EVERY unit the request opens, empty ones included (spec §4.5; today's `revocation_at_commit()`). Writes are re-checked automatically; this is for a non-write route with an explicit commit obligation. ### `authenticates` ``` def authenticates( fn: F | None = None, *, kind: Literal['user', 'service'] = 'user', ) -> F | Callable[[F], F] ``` Mark `fn` (a FastAPI dependency) as authenticating the caller. `@authenticates` (a user subject) or `@authenticates(kind="service")` (a system credential). Deny-by-default (ADR-0009, SANDLOOM-B008): every API route resolves an authenticator or declares `public(reason=...)`; a `module_router()` route with a service authenticator gets no user authentication injected. ### `authorizes` ``` def authorizes(fn: _F) -> _F ``` Mark `fn` (a FastAPI dependency) as a custom authorization check. Recorded, not enforced: the route snapshot lists the marked dependencies a route resolves (`authz`, ADR-0009 "The route snapshot"), so dropping a hand-rolled check shows up as a snapshot diff (SANDLOOM-T001). It neither authenticates (`@authenticates` does) nor carries a risk (`sensitive(...)` does). ### `Capability` ``` def Capability( capability: CapabilityRef[Any] | str, name: str = 'default', *, features: set[str] | None = None, ) -> Any ``` Annotated dependency resolving any capability instance. The universal alternative to a bespoke dependency per capability: it resolves through the request app's core, enforcing any required features, and the version when the reference carries a `.vN` suffix. A capability's package publishes the reference and the Protocol: ``` from vector_index.contracts import VECTOR_INDEX, VectorIndex async def handler( index: Annotated[VectorIndex, Capability(VECTOR_INDEX, "products")], ): ... ``` ### `current_service` ``` async def current_service(x_internal_token: str | None = Header(None)) -> UserContext ``` The service principal of an internal call (`SOURCE_INTERNAL`): the internal token authenticates it (404 otherwise, existence hidden). A route taking it is service-authenticated, never user-authenticated (spec §2.3). ### `current_user_id` ``` async def current_user_id( subject: sandloom.shared.subject.UserContext = Depends(dependency=, use_cache=True, scope=None), ) -> uuid.UUID ``` Extract the authenticated user id from the Bearer access token. ### `CurrentService` ``` CurrentService = Annotated[sandloom.shared.subject.UserContext, Depends(dependency=, use_cache=True, scope=None)] ``` *(no docstring)* ### `Database` ``` def Database(name: str) -> Any ``` Annotated dependency for a named database session. Usage in an endpoint signature: ``` async def handler(db: Annotated[AsyncSession, Database("analytics")]): ... ``` ### `DbSession` ``` DbSession = Annotated[sqlalchemy.ext.asyncio.session.AsyncSession, Depends(dependency=, use_cache=True, scope='function')] ``` *(no docstring)* ### `external_call` ``` def external_call(name: str) -> AsyncIterator[None] ``` Run one external (PSP) call with no unit of the request open (§4.6). Refused (SANDLOOM-R008) while a unit of the request is open; a held contract read-scope connection is released first. While it is active no unit, contract call or connection checkout may start in the request. ### `get_client_ip` ``` def get_client_ip(request: Any, *, config: Any | None = None) -> str ``` Return the real client IP for `request` (Starlette `Request`-like). Only needs `request.client.host` and `request.headers.get`, so it also works for lightweight stand-ins in tests. `config` overrides the settings object (defaults to the process settings). ### `idempotency_guard` ``` def idempotency_guard( *, scope: str, ttl_seconds: int = 86400, required: bool = True, fail_closed: bool = False, store_response: bool = True, durable: bool = False, ) -> Any ``` Public re-export of the Idempotency-Key dependency factory. See `sandloom.core.idempotency`. Use as an endpoint dependency to make a mutating operation safe against duplicate client retries. Pass `fail_closed=True` for money / irreversible side effects so a cache outage returns 503 instead of allowing a possible duplicate. `store_response=False`: in-flight dedupe only — nothing is stored or replayed (the endpoint owns durable idempotency; its response must not be replayed without re-authorization). `durable=True`: the record is claimed in the request's `DbSession` Unit of Work and committed with the endpoint's effect (exactly-once effect on that database, replay of the stored response; ADR-0005) — the form a money route that stores responses needs (SANDLOOM-B019). ### `ModuleRouter` ``` class ModuleRouter(APIRouter)( *, sandloom_owner: str, sandloom_tenant: TenantBinding = 'derive', prefix: str = '', tags: list[str | Enum] | None = None, dependencies: Sequence[params.Depends] | None = None, **kwargs: Any, ) -> None ``` An `APIRouter` whose route decorators also take `tenant: Literal["required", "optional", "none"] | None = None`. Built by `module_router`; collected by discovery wherever it is defined in the module's `api` package. #### `ModuleRouter.route_policy` ``` def route_policy( self, path: str, endpoint: Callable[..., Any], dependencies: Sequence[params.Depends] | None, tenant: RouteTenant, ) -> list[params.Depends] ``` The dependencies `add_api_route` adds in front of the route's. #### `ModuleRouter.add_api_route` ``` def add_api_route( self, path: str, endpoint: Callable[..., Any], *, tenant: RouteTenant = None, dependencies: Sequence[params.Depends] | None = None, **kwargs: Any, ) -> None ``` #### `ModuleRouter.api_route` ``` def api_route( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` #### `ModuleRouter.get` ``` def get( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP GET operation. ## Example \`\`python from fastapi import APIRouter, FastAPI app = FastAPI() router = APIRouter() @router.get("/items/") def read\_items(): return \[{"name": "Empanada"}, {"name": "Arepa"}\] app.include\_router(router) \`\` #### `ModuleRouter.put` ``` def put( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP PUT operation. ## Example \`\`python from fastapi import APIRouter, FastAPI from pydantic import BaseModel class Item(BaseModel): name: str description: str | None = None app = FastAPI() router = APIRouter() @router.put("/items/{item\_id}") def replace\_item(item\_id: str, item: Item): return {"message": "Item replaced", "id": item\_id} app.include\_router(router) \`\` #### `ModuleRouter.post` ``` def post( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP POST operation. ## Example \`\`python from fastapi import APIRouter, FastAPI from pydantic import BaseModel class Item(BaseModel): name: str description: str | None = None app = FastAPI() router = APIRouter() @router.post("/items/") def create\_item(item: Item): return {"message": "Item created"} app.include\_router(router) \`\` #### `ModuleRouter.delete` ``` def delete( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP DELETE operation. ## Example \`\`python from fastapi import APIRouter, FastAPI app = FastAPI() router = APIRouter() @router.delete("/items/{item\_id}") def delete\_item(item\_id: str): return {"message": "Item deleted"} app.include\_router(router) \`\` #### `ModuleRouter.options` ``` def options( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP OPTIONS operation. ## Example \`\`python from fastapi import APIRouter, FastAPI app = FastAPI() router = APIRouter() @router.options("/items/") def get\_item\_options(): return {"additions": \["Aji", "Guacamole"\]} app.include\_router(router) \`\` #### `ModuleRouter.head` ``` def head( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP HEAD operation. ## Example \`\`python from fastapi import APIRouter, FastAPI from pydantic import BaseModel class Item(BaseModel): name: str description: str | None = None app = FastAPI() router = APIRouter() @router.head("/items/", status\_code=204) def get\_items\_headers(response: Response): response.headers\["X-Cat-Dog"\] = "Alone in the world" app.include\_router(router) \`\` #### `ModuleRouter.patch` ``` def patch( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP PATCH operation. ## Example \`\`python from fastapi import APIRouter, FastAPI from pydantic import BaseModel class Item(BaseModel): name: str description: str | None = None app = FastAPI() router = APIRouter() @router.patch("/items/") def update\_item(item: Item): return {"message": "Item updated in place"} app.include\_router(router) \`\` #### `ModuleRouter.trace` ``` def trace( self, path: str, *, tenant: RouteTenant = None, **kwargs: Any, ) -> Callable[[DecoratedCallable], DecoratedCallable] ``` Add a *path operation* using an HTTP TRACE operation. ## Example \`\`python from fastapi import APIRouter, FastAPI from pydantic import BaseModel class Item(BaseModel): name: str description: str | None = None app = FastAPI() router = APIRouter() @router.trace("/items/{item\_id}") def trace\_item(item\_id: str): return None app.include\_router(router) \`\` ### `no_idempotency` ``` def no_idempotency(*, reason: str) -> params.Depends ``` Opt a write route out of the automatic idempotency guard (streaming, non-JSON bodies). The reason is recorded in the route snapshot. ### `operation_intent` ``` def operation_intent(*, reason: str) -> params.Depends ``` Declare the operation-intent mode (spec §4.6): no request unit, sequential `open_uow()` units, external calls in `external_call()`, and `release_check()` before a 2xx (SANDLOOM-R008 / SANDLOOM-R012). Only for routes that call an external provider (billing checkout and portal). ### `rate_limiter` ``` def rate_limiter(*, requests: int, period: int, fail_closed: bool = False) -> Any ``` Expose rate limiter dependency routed through sandloom.core.ingress.rate\_limit. Import lazily so test patches against `sandloom.core.ingress.rate_limit.rate_limiter` are respected. Pass `fail_closed=True` on admission-control endpoints (login, password reset, payment) so a cache outage returns 503 rather than silently lifting the limit. ### `ReadOnlyDbSession` ``` ReadOnlyDbSession = Annotated[sqlalchemy.ext.asyncio.session.AsyncSession, Depends(dependency=, use_cache=True, scope='function')] ``` *(no docstring)* ### `release_check` ``` async def release_check( account: ResolvedAccount, *, authority: Callable[[], Awaitable[None]] | None = None, ) -> None ``` The final re-read before an intent route releases a bearer value. `authority` (the authority re-read, e.g. billing's `reverify_authority`) is REQUIRED for an organization account (SANDLOOM-R012). Then the request subject's session is checked live (401 revoked / blocked; any error fails closed). For its whole duration no unit, command or connection checkout may start in the request, except the authority callback's own read-contract calls (SANDLOOM-R008). ### `replay_safe` ``` def replay_safe(*, reason: str) -> params.Depends ``` Opt a keyed write in to BODY replay (idempotency mode B, spec §4.7). Only when the response contains nothing whose visibility could change between the attempts (typically a creation echoing the caller's own fields and the new id) — never a secret, a bearer capability, other users' data or role / ownership dependent fields. The reason is reviewed in the route snapshot. Without it the automatic guard replays status only. ### `request_operation_key` ``` def request_operation_key() -> str | None ``` The request's operation key: a stable digest of its route, caller, tenant and `Idempotency-Key` as claimed by the route's idempotency guard, or `None` (no key, or no guard). Pass it as a money effect's `operation_key` (ADR-0009 item 11). ### `require_platform_admin` ``` async def require_platform_admin( subject: sandloom.shared.subject.UserContext = Depends(dependency=, use_cache=True, scope=None), ) -> None ``` *(no docstring)* ### `ResolvedAccount` ``` class ResolvedAccount(kind: Literal['user', 'org'], id: UUID) -> None ``` The billing account a release check is for (spec §2.3, §4.6, §7). ### `sensitive` ``` def sensitive(risk: Risk) -> params.Depends ``` The escape hatch for a route whose authority check is hand-rolled: declare its risk (`admin` / `money` / `security`) so the framework applies what that risk triggers (ADR-0009 item 9). ### `TenantId` ``` TenantId = Annotated[uuid.UUID, Depends(dependency=._dep>, use_cache=True, scope=None)] ``` *(no docstring)* ### `TenantScope` ``` def TenantScope( *, header: str = 'X-Tenant-Id', claim_keys: tuple[str, ...] = ('tenant_id', 'org_id'), required: bool = True, ) -> Any ``` FastAPI dependency that binds the request's tenant for its whole scope. Resolves the tenant id from the `X-Tenant-Id` header, else a JWT claim (`tenant_id`/`org_id`). Binds the contextvar before the endpoint runs and resets it afterwards, so `scoped()` / automatic enforcement see it. With `required` (default) a missing tenant is a 400; otherwise it yields `None` and leaves the context unbound. Usage: ``` async def handler(tenant: Annotated[UUID, TenantScope()]): ... ``` ### `writes` ``` def writes(database: str) -> params.Depends ``` Name the route's write store when it injects several writable sessions (spec §6.1, SANDLOOM-B009). ## Examples and related guides - [Route policy]() - [Python API overview]()