sandloom.sdk.data¶

Tables, sessions and reads: Base, TenantScoped, new_id, CreatedAt / UpdatedAt, db(), RLS helpers, the read planner.

Public API 1.0 (spec §2.2, Appendix C).

from sandloom.sdk.data 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¶

Base¶

class Base(DeclarativeBase)(**kwargs: Any) -> None

Base class for all SQLAlchemy models.

CreatedAt¶

CreatedAt = Annotated[datetime.datetime, <sqlalchemy.orm.properties.MappedColumn object>]

(no docstring)

db¶

def db() -> AsyncSession

The session of the innermost open unit of the current task (§4.3):

item = Item(title=body.title)
db().add(item)
await db().flush()   # never commit(): the unit owns the transaction

Outside any unit it raises UnitOfWorkRequiredError (SANDLOOM-R002); in a request the first call opens the lazy request unit, committed when the handler succeeds and rolled back on an exception. A child task opens its own unit with open_uow (SANDLOOM-R013 otherwise).

new_id¶

def new_id() -> uuid.UUID

A new time-ordered id (UUIDv7, RFC 9562) for a primary key:

id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=new_id)

Ids sort in creation order: by time across processes, and strictly increasing within one process (a counter orders a millisecond's burst). So order_by(Model.id) is the creation order, with no tie.

TenantScoped¶

class TenantScoped

Mixin marking a model as tenant-owned (adds the indexed tenant_id).

Inheriting this both declares the column and opts the model into automatic tenant filtering. Declare models as class Foo(TenantScoped, Base).

Advanced names¶

disable_tenant_rls¶

def disable_tenant_rls(op: _Executes, table: str, *, schema: str | None = None) -> None

Drop the Sandloom tenant policies and disable RLS on table (for downgrade).

enable_tenant_rls¶

def enable_tenant_rls(
    op: _Executes,
    table: str,
    *,
    column: str = 'tenant_id',
    schema: str | None = None,
    system_access: bool = False,
    system_role: str = 'sandloom_system',
) -> None

Enable + FORCE PostgreSQL RLS with the Sandloom tenant policy on table.

op is the Alembic op module (or any object with execute(str)). column must be the table's uuid tenant column (TenantScoped declares tenant_id). system_access=True lets the system-channel role (system_role, default sandloom_system; it must already exist) read and write every tenant's rows of THIS table. Re-running is safe (policies are replaced).

get_planned_reads¶

def get_planned_reads(request: Request) -> AsyncGenerator[Any, None]

The handle of a route DECLARED read-only (execution planner).

The whole handler runs inside a read-only scope: publish_event / record_audit / on_commit raise. Declared operations run through reads.one(op, **key) on the planner's general or optimized executor.

get_read_only_db¶

def get_read_only_db(request: Request) -> AsyncGenerator[AsyncSession, None]

PROTOTYPE: request session for a route DECLARED read-only.

With DB_READ_ONLY_SESSIONS=true statements run in a server-enforced READ ONLY transaction. The client policy also refuses writes, flushes, locking reads and raw statements before sending them. PostgreSQL remains the backstop for SELECT functions with side effects. Off (default): exactly get_db.

get_snapshot_reads¶

def get_snapshot_reads(request: Request) -> AsyncGenerator[Any, None]

PlannedReads with ONE snapshot: REPEATABLE READ READ ONLY.

Every declared read of the route runs in the route's read unit, in one REPEATABLE READ read-only transaction (the default PlannedReads is READ COMMITTED: two reads may see different snapshots). Use it when the route's reads must be mutually consistent.

get_uow_reads¶

async def get_uow_reads(request: Request, db: DbSession) -> Any

Planned reads INSIDE the request Unit of Work (mixed routes).

Depends on DbSession (the SAME session the route gets: FastAPI caches the dependency per request), so declared operations run on the UoW's connection and transaction with the general executor: one connection, one tenant GUC, read-your-own-writes, and the UoW's outbox / audit / on_commit semantics unchanged (no read-only scope is opened).

InvalidCursor¶

class InvalidCursor(ValueError)(detail: str = 'invalid pagination cursor') -> None

A pagination cursor that is malformed, tampered with or foreign (400).

JoinRead¶

class JoinRead(ReadOperation)(
    entity: type,
    result: type[R],
    by: tuple[str, ...] = ('id',),
    tenant_scoped: bool | None = None,
    database: str = 'default',
    name: str = '',
    lock: str | None = None,
    join: type | None = None,
    via: str = '',
    join_by: Mapping[str, str] = <factory>,
    join_where: Mapping[str, Any] = <factory>,
    joined_fields: Mapping[str, str] = <factory>,
) -> None

A declared single-row read of entity joined to join on a FK.

entity JOIN join ON entity.<via> = join's primary key, with

• by — equality key columns of entity (call arguments); they must cover entity's primary key or a unique constraint, so the read returns at most one row on EVERY executor. via counts towards that key when the join side is pinned to at most one row (join_by + join_where cover a unique key of join): a read keyed only on the joined row (by=()) is allowed when via is itself unique on entity (e.g. a one-to-one detail row of a customer found by its unique tenant id);
• join_by — {argument: column of join}: more equality arguments;
• join_where — {column of join: constant}: fixed equality filters (e.g. {"is_active": True}), sent as parameters;
• joined_fields — {DTO field: attribute of join}; every other DTO field is read from entity.

via must be a column of entity with a declared ForeignKey to join (no guessed join conditions). Tenant scoping is inferred PER SIDE: every TenantScoped table of the join gets the tenant predicate (and RLS). Both entities must belong to the declaring module (the import boundary rules already forbid reaching another module's models).

ListOperation¶

class ListOperation(ReadOperation)(
    entity: type,
    result: type[R],
    by: tuple[str, ...] = (),
    tenant_scoped: bool | None = None,
    database: str = 'default',
    name: str = '',
    lock: str | None = None,
    where: Mapping[str, Any] = <factory>,
    order_by: tuple[str, ...] = ('id',),
    descending: bool = False,
    max_limit: int | None = None,
    require_index: bool = True,
    join: type | None = None,
    via: str = '',
    join_by: Mapping[str, str] = <factory>,
    join_where: Mapping[str, Any] = <factory>,
    joined_fields: Mapping[str, str] = <factory>,
    optional: Mapping[str, str] = <factory>,
    prefix: Mapping[str, str] = <factory>,
    prefix_ignore_case: bool = False,
) -> None

A declared list read: entity rows, one keyset page per call.

• by — equality filter columns of entity (call arguments; may be empty for a tenant-scoped "all rows of my tenant" list);
• where — {column of entity: constant} fixed filters;
• order_by — columns of entity, each part of an index of its table; the primary key is appended as the tie-breaker (a total order, so keyset pages never skip or repeat a row); descending flips the whole order;
• max_limit — the declaration's page cap (never above PLANNER_LIST_MAX_LIMIT); every call passes limit (mandatory);
• optional join on a DECLARED foreign key, in either direction: via on entity -> join's primary key (many-to-one), or via on join -> entity's primary key (one-to-many: via + join_by + join_where must cover a unique key of join, so the join never multiplies rows); join_by / join_where / joined_fields as for JoinRead;
• optional — {argument: column of entity} equality filters applied only when the call's argument is not None (None = no filter, never IS NULL);
• prefix — {argument: String column of entity}: column LIKE <argument> || '%' with the argument matched as a LITERAL prefix (its % / _ / \ are escaped), applied only when the argument is not None; prefix_ignore_case=True matches case-insensitively (ILIKE).

Optional and prefix filters are residual predicates: the covering-index rule is about by + where (+ tenant) and the order only — a page walks that index in order and skips non-matching rows, it never sorts. Each combination of present filters is its own statement (at most MAX_OPTIONAL_FILTERS optional + prefix filters per operation).

Pages are addressed by an opaque, tamper-evident cursor (HMAC-SHA256 over the operation, the filter arguments, the bound tenant and the position): the client cannot move it to another filter, tenant or position. The tenant is never read from it. Tenant scoping is inferred per side.

ListOperation.n_required¶

n_required  (property)

The number of leading keys that are always applied.

ListOperation.page_cap¶

def page_cap(self) -> int

The effective page cap: the declaration's, never above the framework's.

make_cursor¶

def make_cursor(op: ListOperation[Any], position: Sequence[Any], /, **key: Any) -> str

A cursor starting after position (the operation's order values, in order_by + primary-key order) for these filter arguments and the bound tenant (e.g. to resume a list at a known row).

make_db_dependency¶

def make_db_dependency(
    name: str,
) -> Callable[[Request], AsyncGenerator[AsyncSession, None]]

Build a request dependency yielding a session for a NAMED database.

Use this when a module's endpoints must read/write a non-default database (e.g. "analytics"). The module references only the logical name; the concrete engine/driver is resolved by the infrastructure registry. Events published while this session is the active UoW go to that database's transactional outbox.

DbSession remains the canonical dependency for the default database.

MoneyLedger¶

class MoneyLedger

Model mixin of a money ledger: names its unique operation-key column.

open_uow¶

def open_uow(database: str | None = None) -> AbstractAsyncContextManager[AsyncSession]

Open a unit of work on database (today db_session, §4.3):

async with open_uow() as session:
    session.add(row)          # committed when the block exits

One transaction, committed on success and rolled back on an exception, with the transactional outbox for the events it publishes.

Page¶

class Page(Generic)(items: list[R], next_cursor: str | None) -> None

One page of a ListOperation: the DTOs and the next cursor.

PageLimitError¶

class PageLimitError(ValueError)(detail: str) -> None

limit missing, < 1 or above the operation's page cap (400).

PlannedReads¶

PlannedReads = Annotated[Any, Depends(dependency=<function get_planned_reads>, use_cache=True, scope='function')]

(no docstring)

ReadOperation¶

class ReadOperation(Generic)(
    entity: type,
    result: type[R],
    by: tuple[str, ...] = ('id',),
    tenant_scoped: bool | None = None,
    database: str = 'default',
    name: str = '',
    lock: str | None = None,
) -> None

A declared single-row read: entity WHERE by columns = arguments.

result is part of the contract: the caller always gets an instance of it (or None), whichever executor ran. tenant_scoped is inferred from the entity; declaring it explicitly is checked against the entity (an operation can never opt OUT of tenant scoping of a tenant-owned entity). Identity-hashed: one object = one declared operation.

Reads¶

class Reads(
    core: CoreRuntime | None,
    unit: Any = None,
    *,
    adapter: ExecutionAdapter | None = None,
    shared: bool = False,
    contract: bool = False,
    isolation: Isolation = 'read_committed',
) -> None

Handle that runs declared operations.

Two modes, fixed when the handle is made:

• read-only scope (unit is None; sandloom.core.di.PlannedReads, read_scope): consecutive OPTIMIZED reads share a lazy read-only transaction per database. A GENERAL read releases that transaction and opens an ORM read unit; subsequent reads use that unit. Stronger isolation always uses one ORM transaction. A scope never holds two connections of one pool. A handle made without a scope (shared=False, execute_one) runs every operation in its own transaction;
• inside a Unit of Work (unit = the UoW's session; sandloom.core.di.UowReads, sandloom.core.orm.session_reads): every operation runs ON that session — the UoW's connection, transaction and tenant GUC — with the GENERAL executor (ORM: autoflush + identity map, so the read sees the unit's own writes, flushed or not). The optimized executor is never used there (see the design doc, "mixed operations").

Reads.in_unit_of_work¶

in_unit_of_work  (property)

Reads.one¶

async def one(self, op: ReadOperation[R], /, **key: Any) -> R | None

Run op for key (the hot path: one adapter lookup, no wrappers).

Reads.page¶

async def page(
    self,
    op: ListOperation[R],
    /,
    *,
    limit: int,
    cursor: str | None = None,
    **key: Any,
) -> Page[R]

One keyset page of op for the filter arguments key.

limit is mandatory and capped (ListOperation.page_cap, PageLimitError); cursor is the previous page's next_cursor and is verified BEFORE any I/O (InvalidCursor).

Reads.aclose¶

async def aclose(self) -> None

End the scope's read units (the scope owner calls it, once).

session_reads¶

def session_reads(session: Any) -> sandloom.core.planner.Reads

Run declared operations ON session (an open Unit of Work).

Inside a DbSession route, db_session() or contract_session(): the reads share the unit's transaction, connection and tenant context (read-your-own-writes; the general executor). Routes use sandloom.core.di.UowReads instead.

SnapshotReads¶

SnapshotReads = Annotated[Any, Depends(dependency=<function get_snapshot_reads>, use_cache=True, scope='function')]

(no docstring)

UowReads¶

UowReads = Annotated[Any, Depends(dependency=<function get_uow_reads>, use_cache=True, scope='function')]

(no docstring)

UpdatedAt¶

UpdatedAt = Annotated[datetime.datetime, <sqlalchemy.orm.properties.MappedColumn object>]

(no docstring)

Examples and related guides¶

• Units and transactions
• Python API overview