# `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 `. 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 | | --- | --- | --- | | [`Base`](<#base>) | class | Base class for all SQLAlchemy models. | | [`CreatedAt`](<#createdat>) | annotated type | | | [`db`](<#db>) | function | The session of the innermost open unit of the current task (§4.3): | | [`new_id`](<#new_id>) | function | A new time-ordered id (UUIDv7, RFC 9562) for a primary key: | | [`TenantScoped`](<#tenantscoped>) | class | Mixin marking a model as tenant-owned (adds the indexed `tenant_id`). | ## Advanced | Name | Kind | Summary | | --- | --- | --- | | [`disable_tenant_rls`](<#disable_tenant_rls>) | function | Drop the Sandloom tenant policies and disable RLS on `table` (for downgrade). | | [`enable_tenant_rls`](<#enable_tenant_rls>) | function | Enable + FORCE PostgreSQL RLS with the Sandloom tenant policy on `table`. | | [`get_planned_reads`](<#get_planned_reads>) | function | The handle of a route DECLARED read-only (execution planner). | | [`get_read_only_db`](<#get_read_only_db>) | function | PROTOTYPE: request session for a route DECLARED read-only. | | [`get_snapshot_reads`](<#get_snapshot_reads>) | function | `PlannedReads` with ONE snapshot: `REPEATABLE READ READ ONLY`. | | [`get_uow_reads`](<#get_uow_reads>) | async function | Planned reads INSIDE the request Unit of Work (mixed routes). | | [`InvalidCursor`](<#invalidcursor>) | exception | A pagination cursor that is malformed, tampered with or foreign (400). | | [`JoinRead`](<#joinread>) | class | A declared single-row read of `entity` joined to `join` on a FK. | | [`ListOperation`](<#listoperation>) | class | A declared list read: `entity` rows, one keyset page per call. | | [`make_cursor`](<#make_cursor>) | function | A cursor starting after `position` (the operation's order values, in `order_by` + primary-key order) for these filter arguments and the b... | | [`make_db_dependency`](<#make_db_dependency>) | callable | Build a request dependency yielding a session for a NAMED database. | | [`MoneyLedger`](<#moneyledger>) | class | Model mixin of a money ledger: names its unique operation-key column. | | [`open_uow`](<#open_uow>) | function | Open a unit of work on `database` (today `db_session`, §4.3): | | [`Page`](<#page>) | class | One page of a `ListOperation`: the DTOs and the next cursor. | | [`PageLimitError`](<#pagelimiterror>) | exception | `limit` missing, \< 1 or above the operation's page cap (400). | | [`PlannedReads`](<#plannedreads>) | annotated type | | | [`ReadOperation`](<#readoperation>) | class | A declared single-row read: `entity` WHERE `by` columns = arguments. | | [`Reads`](<#reads>) | class | Handle that runs declared operations. | | [`session_reads`](<#session_reads>) | function | Run declared operations ON `session` (an open Unit of Work). | | [`SnapshotReads`](<#snapshotreads>) | annotated type | | | [`UowReads`](<#uowreads>) | annotated type | | | [`UpdatedAt`](<#updatedat>) | annotated type | | ## Beginner names ### `Base` ``` class Base(DeclarativeBase)(**kwargs: Any) -> None ``` Base class for all SQLAlchemy models. ### `CreatedAt` ``` CreatedAt = Annotated[datetime.datetime, ] ``` *(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] = , join_where: Mapping[str, Any] = , joined_fields: Mapping[str, str] = , ) -> None ``` A declared single-row read of `entity` joined to `join` on a FK. `entity` JOIN `join` ON `entity.` = `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] = , 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] = , join_where: Mapping[str, Any] = , joined_fields: Mapping[str, str] = , optional: Mapping[str, str] = , prefix: Mapping[str, str] = , 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=, 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=, use_cache=True, scope='function')] ``` *(no docstring)* ### `UowReads` ``` UowReads = Annotated[Any, Depends(dependency=, use_cache=True, scope='function')] ``` *(no docstring)* ### `UpdatedAt` ``` UpdatedAt = Annotated[datetime.datetime, ] ``` *(no docstring)* ## Examples and related guides - [Units and transactions]() - [Python API overview]()