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

CurrentUser¶

CurrentUser = Annotated[sandloom.shared.subject.UserContext, Depends(dependency=<function current_context>, 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 <module>/api.py:

router = module_router()

@router.post("/", status_code=201)
async def create_note(body: NoteIn, user: CurrentUser) -> NoteOut: ...

prefix defaults to /api/v1/<module> and tags to [<module>]. 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=<function current_context>, 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=<function current_service>, 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=<function get_db>, 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=<function get_read_only_db>, 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=<function current_context>, 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=<function TenantScope.<locals>._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