Data and analysis#

Wreath owns its PostgreSQL driver and model stack. It also owns the bounded analytical and storage primitives that sit beside transactional data.

wreath.postgres#

PostgreSQL driver facade and application-owned workload pools.

AdvisoryLock#

A blocking, session-scoped advisory lock held for an async with block.

The lock pins a single connection from workload (default "write", i.e. the primary) for the lifetime of the block and releases it -- and the lock -- on exit. Because a session-scoped lock is bound to its backend connection, the connection is withheld from the pool while held; hold enough of these concurrently and ordinary queries can starve on acquire. Prefer the xact-scoped Session.lock for request-path exclusion and reserve this for a handful of long-lived fleet locks.

AdvisoryLock(database: Database, key: str, *, namespace: str | None = None, mode: str = 'exclusive', workload: Workload = 'write') -> None

AdvisoryTryLock#

A non-blocking (or timeout-bounded) session-scoped advisory lock.

Used as async with db.try_lock(key) as held: -- held is the lock handle when acquired, or None when it could not be taken. With timeout set, acquisition blocks up to that many seconds using PostgreSQL's lock_timeout (which inherits the server's fair-ish lock queue), not a client-side spin.

AdvisoryTryLock(database: Database, key: str, *, timeout: float | None = None, namespace: str | None = None, mode: str = 'exclusive', workload: Workload = 'write') -> None

Connection#

Bounded automatically-pipelined asynchronous PostgreSQL connection.

close (inherited from Connection)#

async close() -> None

closed (property, inherited from Connection)#

closed: bool

dropped_notifications (property, inherited from Connection)#

dropped_notifications: int

Async notifications discarded because the ring overflowed.

execute (inherited from Connection)#

async execute(sql: str, *args: object) -> str

fetch (inherited from Connection)#

async fetch(sql: str, *args: object) -> list[Record]

fetch_batch (inherited from Connection)#

async fetch_batch(sql: str, *args: object) -> RecordBatch

fetchrow (inherited from Connection)#

async fetchrow(sql: str, *args: object) -> Record | None

fetchval (inherited from Connection)#

async fetchval(sql: str, *args: object) -> object

listen (inherited from Connection)#

async listen(channel: str) -> None

Subscribe this connection to asynchronous NOTIFY on channel.

Issues LISTEN and registers interest so the reader task stays alive while idle and routes incoming notifications into the ring. Notifications are delivered through notifications(). A dedicated connection should be held for listening: leasing it back to a pool between notifications would stop draining them (that lifecycle is the caller's concern).

map (inherited from Connection)#

async map(method: ResultMode, statement: object, argument_sets: Any, *, max_in_flight: int = 32) -> list[Any]

Run one method operation per argument set, preserving input order.

Every input produces a distinct extended-protocol operation with its own Sync; duplicate inputs run twice and are never coalesced into an IN (...) or deduplicated. Results come back in input order. At most max_in_flight operations are queued at once, so a generator input keeps pipeline depth and retained results bounded. Inside an explicit transaction the driver already forbids concurrent operations, so the window collapses to one and the operations run strictly in order.

statement may be a SQL string or any object exposing .sql.

notifications (inherited from Connection)#

notifications() -> AsyncIterator[Notification]

Yield captured NOTIFY messages, awaiting new ones when drained.

The iterator ends when the connection closes. It never blocks while the ring is non-empty, so a burst is delivered without interleaved awaits.

prepared_plan_count (property, inherited from Connection)#

prepared_plan_count: int

Automatic prepared plans currently cached on this connection.

transaction (inherited from Connection)#

transaction() -> _Transaction

An explicit transaction scope over this connection.

async with connection.transaction() as tx: issues BEGIN on entry and COMMIT on clean exit, or ROLLBACK if the body raises. Reads and writes issued through tx run in order, so a read group completes before a dependent write group begins. If synchronization cannot be proven on rollback, the connection is discarded rather than reused.

unlisten (inherited from Connection)#

async unlisten(channel: str) -> None

Stop listening on channel (issues UNLISTEN).

The channel is deregistered immediately; if it was the last one the idle reader may stay parked on the socket until the next message or close() before it observes the empty channel set and exits.

Database#

One logical database with independently bounded workload pools.

Database(name: str, dsn: str, *, pools: Mapping[Workload, PoolConfig | Mapping[str, Any]] | None = None, workload_dsns: Mapping[Workload, str] | None = None, connector: Connector | None = None, shutdown_timeout: float = 10.0) -> None

acquire#

async acquire(workload: Workload = 'read') -> Any

Lease a connection from workload's pool. Pair it with release.

The one acquisition seam in this module: Statement, Statement.map and direct callers all come through here, which is why the Flight Recorder's pool-wait phase can be attributed from this single point.

Raises whatever Pool.acquire raises, plus KeyError when the workload is not configured.

acquire_shared#

async acquire_shared(workload: Workload = 'read') -> Any

Lease a connection concurrent statements may share. See PoolConfig.

Additive and internal on purpose. Folding shared= into acquire() instead looked tidier -- one seam, one keyword -- and broke twenty-four tests across jobs, passes and workflow recording, because every Database-shaped double in the tree is written against acquire()'s current signature. A new method they do not have is invisible to them; a new keyword on one they do have is not.

Statement reaches this through _acquire_for_statement, which falls back to acquire() for any database object that does not implement it -- so a double keeps working and simply gets exclusive leases, which is the conservative answer rather than a failure.

counters#

counters() -> Any

This database's pools, summed, for wreath.metrics.collect.

One reading per database rather than per workload pool: the instance label is the database, and a caller who wants the split reads pool(workload).snapshot(). Summing is right for the four fields that are counts of connections; queue_high_water is a maximum, so it is taken as one rather than added.

Only pools that exist are read. Asking for one that has not been built would create it, and a metrics scrape must not open connections.

lock#

lock(key: str, *, namespace: str | None = None, mode: str = 'exclusive', workload: Workload = 'write') -> AdvisoryLock

A blocking, session-scoped advisory lock held across an async with.

The lock pins one connection from workload (the primary by default) for the block's duration. Prefer Session.lock(scope="xact") for request-path exclusion; use this for long-lived fleet locks.

name (property)#

name: str

The logical name this database was registered under.

The handle an application knows it by — the name given to Wreath.postgres, which is also what Wreath.orm(database=...) and the other database-taking helpers name — rather than anything PostgreSQL itself knows. Required, non-empty, and fixed at construction.

pool#

pool(workload: Workload) -> Pool

The pool serving workload.

Available only once the database has started. Before that, the call exists to answer one question — is this workload configured at all? — and answers it by raising one of two different errors, which are worth telling apart.

Raises:

• KeyError — this workload is not configured on this database, whether
• or not it has started.
• InterfaceError — the workload is configured, but the database has not
• started, so there is no pool to hand back yet. Inspecting a pool
• before startup is deliberately unsupported.

release#

async release(workload: Workload, connection: Any) -> None

Return a connection to the pool it was leased from.

workload must be the one it was acquired with — the connection is handed to that pool, and a pool refuses a connection it did not lease with InterfaceError.

release_shared#

async release_shared(workload: Workload, connection: Any) -> None

Return a shared lease to the pool it came from.

run_singleton#

run_singleton(key: str, work: Callable[[], Awaitable[Any]], *, namespace: str | None = None, workload: Workload = 'write', poll_interval: float = 5.0) -> SingletonRunner

Run work once across the fleet, guarded by an advisory lock.

work is a zero-argument callable returning a fresh awaitable. Returns a handle with await handle.stop(); wire it through on_startup / on_shutdown. The guarded critical section must be idempotent.

start#

async start() -> None

Build a Pool per configured workload and start them all.

Each pool gets the workload's own DSN when one was supplied and the database's otherwise, opens read-only for read and security_read, and prepares that workload's registered statements on every connection.

All-or-nothing across pools as well as within them: if one fails to start, the pools already up are stopped, the set is cleared, and the exception propagates with started still false. Calling it twice does nothing the second time.

A workload first named by statement() once this call has begun is not started here, and pool() raises KeyError for it.

statement#

statement(name: str, sql: str, *, workload: Workload = 'read') -> Statement

Register one prepared statement, refusing a name already claimed.

The duplicate check and the assignment run under a lock because separately they are not a check at all: two threads registering the same name both passed name in self._statements before either assigned, so both succeeded and the loser was left holding a Statement that no pool ever prepares -- _for_workload only ever sees the survivor. A guard that exists to catch two subsystems claiming one name silently became a lost write instead of the refusal it was written to be.

The window spans a Statement construction, so it does not need a free-threaded interpreter to be real; free-threading merely removes the GIL that makes it rare. Registration happens at startup and on a store's first use, never per request, so a plain lock costs nothing worth measuring. store.py's _prepare_lock guards the tier above this one.

stop#

async stop() -> None

Stop every pool, in reverse start order, within one shared deadline.

shutdown_timeout is the budget for the whole shutdown, not for each pool: whatever an earlier pool spends draining is taken off what the next one gets, and a pool reached after the deadline has passed is stopped with no grace at all. Calling it on a database that never started does nothing.

try_acquire_shared#

try_acquire_shared(workload: Workload = 'read') -> Any | None

The uncontended shared lease, taken without a coroutine.

None means it has to be awaited: nothing idle, or the pool is shutting down -- and deliberately also whenever a Flight Recorder phase marker is armed, because acquire_shared is the one seam pool waiting is attributed from and an armed request must keep going through it.

See Pool.try_acquire_shared for why the synchronous path exists.

try_lock#

try_lock(key: str, *, timeout: float | None = None, namespace: str | None = None, mode: str = 'exclusive', workload: Workload = 'write') -> AdvisoryTryLock

A non-blocking advisory lock: async with db.try_lock(k) as held:.

held is the handle when acquired or None otherwise. With timeout set, acquisition blocks up to that many seconds via lock_timeout.

try_release_shared#

try_release_shared(workload: Workload, connection: Any) -> bool

Return a shared lease without a coroutine; False if it has to await.

FromDatabase#

FromDatabase(name: 'str | None' = None, workload: 'Workload' = 'read')

FromDatabase(name: str | None = None, workload: Workload = 'read') -> None

InterfaceError#

Invalid driver use or connection state.

OperationalError#

Transport, authentication, or protocol operation failure.

PipelineFullError#

The connection's bounded pipeline cannot accept another operation.

Pool#

A bounded lease pool for one database workload.

A lease is exclusive by default. PoolConfig.pipeline_depth above 1 lets concurrent statements share a connection so their operations batch into one flight -- see acquire, and the note on that field for the measurement that chose the depth. Anything holding an explicit lease still has the connection to itself, because a shared one cannot carry a transaction.

Pool(dsn: str, config: PoolConfig, *, connector: Connector, read_only: bool, statements: Callable[[], tuple[Statement, ...]]) -> None

acquire#

async acquire(*, shared: bool = False) -> Any

Lease one connection. Always pair it with release.

shared=True allows up to pipeline_depth concurrent holders, so their operations batch into one flight; the default is an exclusive lease.

Three outcomes, tried in this order: an idle connection is handed over immediately; otherwise a new one is opened if the pool is below max_size; otherwise the caller queues until somebody releases one.

Every way of failing is an exception rather than a None, and each says which limit was hit:

Raises:

• InterfaceError — the pool has not started, or is shutting down, or
• the wait queue already holds max_queue callers. A full queue
• refuses immediately rather than growing without bound.
• TimeoutError — acquire_timeout seconds elapsed while queued. The
• deadline covers the wait only — a connection being opened for
• this caller is not interrupted by it.

borrowed (property)#

borrowed: int

How many connections are leased out right now.

A lease is exclusive, so this is also how many are unavailable to the next acquire. snapshot() reports it alongside the idle count, the queue depth and the high-water mark, and is the one to reach for when more than one of them is wanted at a consistent moment.

counters#

counters() -> Any

This pool's counters, for wreath.metrics.collect.

queue_high_water is the one worth scraping: it is the only field that is not instantaneous, so it answers "did anything ever wait?" — which a sampler polling waiters will usually miss entirely.

release#

async release(connection: Any, *, shared: bool = False) -> None

Return a leased connection, and wake one waiter.

shared must match the acquisition: a shared lease is one of several on that connection, and only the last one out returns it to the pool.

The connection goes back to the idle set and is reused — unless the pool is shutting down or the connection reports itself closed, in which case it is dropped from the pool and closed here instead. Either way a waiter is notified, so a connection that turned out to be dead still frees the capacity it was holding.

Raises:

• InterfaceError — this connection was not leased from this pool.
• Releasing twice is the usual way to see it, and it is refused
• rather than tolerated because a double release would put one
• connection in the idle set twice and hand it to two callers.

snapshot#

snapshot() -> PoolSnapshot

Mirrors HTTPClient.snapshot(), so both pools read the same way.

start#

async start() -> None

Open min_size connections and begin accepting acquisitions.

Each connection is opened, set read-only when this pool serves a read workload, and has every statement registered for that workload prepared on it before it joins the idle set — so a Statement never pays a prepare on the request path.

All-or-nothing: if any of them fails, the ones already opened are closed and the exception propagates with the pool still unstarted, rather than leaving a half-sized pool that works until it doesn't. Calling it on an already-started pool does nothing.

started (property)#

started: bool

Whether start() has completed and acquire() will be answered.

False before the first start(), and false again after stop() returns. It does not go false while a stop is draining — use it to ask "has this pool been brought up", not "is it still healthy".

stop#

async stop(grace_period: float) -> None

Stop accepting acquisitions, drain for grace_period, then close everything.

Draining is best-effort: the pool waits up to grace_period seconds for leased connections to come back, and then closes what is still out anyway. A caller still holding one at that point has it closed underneath it — the grace period is the promise, not the outcome.

Waiting callers are woken immediately rather than left on the queue, and acquire refuses from the first moment of the stop. Calling this on a pool that never started does nothing.

Args:

• grace_period — Seconds to wait for leases to be returned. Zero closes
• everything at once.

try_acquire_shared#

try_acquire_shared() -> Any | None

Take an idle connection into the shared set, or return None.

The uncontended shared lease, done in the caller's frame. Returning None means "this one needs the coroutine" -- nothing idle, not started, or shutting down -- and the caller falls through to _acquire_shared, which owns every decision this does not make.

It is a separate synchronous method rather than a branch inside _acquire_shared because the cost being removed is the coroutine: a lease and its release created and stepped seven of them to move one connection between a deque and a dict, which measured 18,732 instructions with no query attached to it. wreath-decomp --suite calibrate prices a non-suspending await at 95.7ns against 49.8ns for a guarded synchronous call, and seven of the first is most of that lease.

The policy stays in one place: this is the same branch _acquire_shared opens with, in the same order, and every other case is still that method's.

try_release_shared#

try_release_shared(connection: Any) -> bool

Give back a shared lease without suspending, or return False.

True means the lease is fully returned. False means the connection has to go back through release, which is every case that can await: the pool is shutting down, or the connection reports itself closed and has to be closed here.

Raises:

• InterfaceError — this connection holds no shared lease from this
• pool. Refused here rather than deferred to the slow path so a
• double release is caught in the frame that made it.

PoolConfig#

PoolConfig(min_size: 'int' = 1, max_size: 'int' = 10, max_queue: 'int' = 256, acquire_timeout: 'float' = 5.0, command_timeout: 'float' = 30.0, statement_cache_size: 'int' = 100, statement_cache_bytes: 'int' = 4194304, pipeline_depth: 'int' = 4)

PoolConfig(min_size: int = 1, max_size: int = 10, max_queue: int = 256, acquire_timeout: float = 5.0, command_timeout: float = 30.0, statement_cache_size: int = 100, statement_cache_bytes: int = 4194304, pipeline_depth: int = 4) -> None

PoolSnapshot#

What the pool looks like right now, for a status page or a pacer.

queue_high_water is the only field that is not instantaneous: it is the deepest the wait queue has ever been, because the interesting question -- "did anything ever wait?" -- has no answer a sampler can be relied on to catch. A pass that paces against pool pressure needs to know the queue was ten deep for fifty milliseconds, and polling waiters will usually miss it.

PoolSnapshot(borrowed: int, available: int, waiters: int, max_size: int, queue_high_water: int) -> None

PostgresError#

An error reported by PostgreSQL.

A subclass that stands for exactly one SQLSTATE declares it as a class attribute; anything raised from a server ErrorResponse passes the code it actually carried. Both spellings answer to .sqlstate.

PostgresError(message: str, *, sqlstate: str | None = None) -> None

ProtocolError#

Malformed or unexpected PostgreSQL wire data.

Record#

RecordBatch#

SingletonRunner#

Run one coroutine at a time across the whole fleet, via an advisory lock.

work is a zero-argument callable returning a fresh awaitable each time it is invoked (a coroutine can only be awaited once, and leadership may be re-established after a failover). The winner holds a dedicated connection and the advisory lock for as long as work() runs; if the process dies, its backend connection drops, PostgreSQL releases the lock, and a follower is promoted within one poll_interval. The guarded work must be idempotent -- failover can hand leadership over mid-flight.

SingletonRunner(database: Database, key: str, work: Callable[[], Awaitable[Any]], *, namespace: str | None = None, workload: Workload = 'write', poll_interval: float = 5.0, jitter: Callable[[float], float] | None = None) -> None

lead_errors (property)#

lead_errors: int

How many leadership attempts ended in a failure rather than a return.

A work() that fails every time is otherwise invisible: leadership is acquired, dropped, and re-contended on a timer, so the fleet looks busy while nothing is being done. This is the only signal that distinguishes "nobody has needed to lead yet" from "leading has never once worked".

release_errors (property)#

release_errors: int

How many rounds ended with the connection's release itself failing.

Distinct from lead_errors, which counts leadership attempts that failed to do their work. This counts rounds whose work may well have succeeded and whose cleanup did not, so a runner that is quietly failing to hand connections back is visible before the pool runs out of them.

stop#

async stop() -> None

Relinquish leadership (if held) and stop contending for it.

SparseVector#

A pgvector sparsevec value: a dimension, and the positions that are not zero.

Construct one from a mapping of 1-based index to value, or from anything dict() accepts (an iterable of pairs, another SparseVector's to_dict()):

SparseVector(5, {1: 1.5, 3: 3.5})       # '{1:1.5,3:3.5}/5' to pgvector
SparseVector.from_dense([1.5, 0, 3.5, 0, 0])   # the same value

Explicit zeros are dropped rather than stored. That is what the server does to them, so keeping them would mean a value did not survive its own round trip -- write {2: 0.0}, read back {} -- and a difference that appears only after a write is the worst kind to debug.

Immutable in practice: indices and values are tuples and there is no setter. Two values are equal when their dimension and their non-zero elements are, so SparseVector(5, {1: 1.0}) equals itself across a round trip through the database.

Attributes: dim: The declared dimension -- how many positions exist, not how many are stored. indices: The stored positions, 1-based and ascending. values: The stored values, positionally paired with indices.

SparseVector(dim: Any, elements: Any = ()) -> None

dim (property)#

dim: int

The declared dimension.

from_dense (classmethod)#

from_dense(values: Any) -> SparseVector

The sparse form of a dense sequence, dropping its zeros.

The dimension is the sequence's length, so a 30,000-word bag whose vocabulary is fixed keeps its dimension even when a document uses nine of the words.

indices (property)#

indices: tuple[int, ...]

The stored 1-based positions, materialized at the Python boundary.

to_dict#

to_dict() -> dict[int, float]

The non-zero elements, 1-based index to value.

values (property)#

values: tuple[float, ...]

The stored values, materialized at the Python boundary.

Statement#

One named, prepared SQL statement bound to a database and a workload.

Register with Database.statement(name, sql, workload=...) rather than constructing directly; that is what makes the pools prepare it on every connection they open, so running it never pays a parse.

A statement carries its workload, and that is the whole of its routing: each call acquires a connection from that pool, runs, and releases it before returning. Nothing is held between calls, so two statements never share a transaction — for that, use a session.

The four run methods differ only in what they ask the connection for, and each takes the query parameters as positional arguments, bound by position ($1, $2, ...). map runs the same statement over many argument sets on one connection.

Attributes: database: The Database this was registered on. name: The registered name, unique within that database. sql: The statement text. workload: Which pool its connections come from — "read", "security_read" or "write". The two read workloads open their connections with default_transaction_read_only.

Statement(database: Database, name: str, sql: str, workload: Workload) -> None

execute#

execute(*args: object) -> Awaitable[str]

Run the statement for its effect and return the command tag.

The tag is PostgreSQL's own summary of what happened — "INSERT 0 1", "UPDATE 3" — so it is where a row count comes from when there are no rows to fetch. Rows the statement does produce are discarded.

fetch#

fetch(*args: object) -> Awaitable[list[Any]]

Run the statement and return every row, as a list of Record.

The whole result is materialized; there is no cursor here. A statement that can match an unbounded number of rows should carry its own LIMIT.

fetch_batch#

fetch_batch(*args: object) -> Awaitable[Any]

Run the statement into a compact row collection.

The batch keeps the familiar sequence and append surface while exposing sort_by(column) so sorting does not allocate a Python key callable or dispatch it once per row. Individual rows remain ordinary Record values when Python observes them.

fetchrow#

fetchrow(*args: object) -> Awaitable[Any]

Run the statement and return its first row, or None for no rows.

None means the query matched nothing. It is not an error, and a statement whose first column can itself be null is better read with this than with fetchval, which cannot tell the two apart.

fetchval#

fetchval(*args: object) -> Awaitable[Any]

Run the statement and return the first column of the first row.

None for no rows and for a first column that is null — the two are indistinguishable here, which is what fetchrow is for.

map#

async map(method: str, argument_sets: Any, *, max_in_flight: int = 32) -> list[Any]

Run this prepared statement once per argument set, in input order.

Acquires one connection for the whole fan-out; each input becomes a distinct Sync-delimited operation (duplicates are not coalesced or deduplicated), and results are returned in input order.

connect#

Open a native PostgreSQL connection.

wreath.orm.model#

Model declaration, object state, and fixed model storage.

ModelMeta compiles a class body into an ordered column layout once, at class creation. Registries later resolve relationships against that layout; no part of the request path inspects annotations or class bodies again.

Every concrete model gets a generated C storage type whose tp_basicsize is fixed, with unboxed inline scalar cells, compact bitmaps, and C descriptors. Assignment validates through the column's PgType.coerce, so declaration and request validation share one type rule.

Model#

Base class for mapped models.

Subclass with table= to map a class to a table:

class User(Model, table="users"):
    id: Mapped[int] = column(Int64, primary_key=True)
    email: Mapped[str] = column(Text, unique=True)

Subclassing without table= declares a reusable mixin whose columns are inherited, in declaration order, ahead of the subclass's own.

Model(**values: Any) -> None

ModelMeta#

Collects ordered column and relationship declarations.

ModelMeta(name: str, bases: tuple[type, ...], namespace: dict[str, Any], **kwargs: Any) -> None

select#

select(*fields: Any) -> Any

Begin a SELECT for this model; no arguments selects every column.

enforce_rules#

enforce_rules(instance: Any) -> None

Raise if any whole-object rule refuses instance.

A CheckViolation is a ValueError, the same thing a refused assignment raises, so a caller that already handles a bad value handles a broken rule without a new except clause.

validate_identifier#

validate_identifier(value: str, kind: str) -> str

Validate ORM's unquoted form through the PostgreSQL name authority.

wreath.orm.fields#

Column declarations and the descriptor that backs field access.

A column() in a class body is a prototype. ModelMeta clones one descriptor per concrete model so a column declared on a shared mixin can hold a different storage index in each model that inherits it.

MISSING (value)#

MISSING: _Missing = MISSING

Column#

A declared column, and the descriptor that reads and writes it.

Column(pg_type: PgType, *, primary_key: bool, nullable: bool, unique: bool, indexed: bool, default: Any, server_default: str | None, references: Any, on_delete: str | None = None, on_update: str | None = None, deferrable: bool = False, checks: tuple[Check, ...] = (), index_method: str | None = None, index_ops: str | None = None, index_with: tuple[tuple[str, str], ...] = ()) -> None

database_name (property)#

database_name: str

expression (property)#

expression: ColumnExpr

generated (property)#

generated: bool

Whether PostgreSQL computes this column rather than the application.

A property rather than a stored flag: it is read on the rare branch of the constructor (a column with no supplied value and no default) and by the migration descriptor, never per assignment -- assignment is refused by the type's own coerce, which is already on that path.

Mapped#

Typing-only marker for a mapped column.

Mapped[int] documents that class access yields a SQL expression and instance access yields an int. It carries no runtime behavior: the registry compiles declarations from column() objects, never from annotations.

column#

column(pg_type: PgType, *, primary_key: bool = False, nullable: bool = False, unique: bool = False, index: bool | str = False, index_ops: str | None = None, index_with: Mapping[str, Any] | None = None, default: Any = MISSING, server_default: str | None = None, references: Any = None, on_delete: str | None = None, on_update: str | None = None, deferrable: bool = False, check: Any = None) -> Any

Declare a mapped column.

default is a Python value or a zero-argument callable applied when a constructor omits the field. index=True declares one ordinary btree index on this column; index="gin" declares a GIN index instead (the right choice for Jsonb and Array columns queried with the containment and key operators), and index="hnsw" or index="ivfflat" declares a pgvector index on a Vector column.

index_ops names the operator class the index is built with, which is what decides which distance an approximate index can answer:

embedding: Mapped[list[float]] = column(
    Vector(1536),
    index="hnsw",
    index_ops="vector_cosine_ops",
    index_with={"m": 16, "ef_construction": 64},
)

index_with passes index-method options through to WITH (m = 16, ef_construction = 64). Both require an index=, and both are rendered into DDL rather than bound, so their values are restricted to plain identifiers and numbers. server_default names a database-side default, which makes the column optional on insert and returned by RETURNING. references takes another model's column expression (references=User.id) and records a foreign key.

check takes one constraint from wreath.orm.constraints, or a sequence of them, applied in order after the type has accepted the value:

salary: Mapped[int] = column(Int64, check=Ge(0))
name: Mapped[str] = column(Text, check=[Length(1, 200), Pattern(r"\S")])

Checks run on every write -- the constructor, assignment, and a request body -- and cost one comparison each, not one call each. A subclass adds to them with narrow(); nothing removes them.

encode_default#

encode_default(value: Any) -> bytes

A deterministic tagged encoding of a column default.

resolve_default#

resolve_default(spec: Any) -> Any

Produce a fresh default value for spec, calling factory defaults.

wreath.orm.constraints#

Business rules, layered onto the column types inside the same single pass.

A column type answers what a value is: an int8 is a 64-bit integer, and nothing else gets in. It cannot answer what a value is allowed to be here: an intern's salary is an int8, but an intern's salary above 50,000 is a business mistake, not a type error.

This module adds that second question without adding a second engine. A check is fused into the column's own coercion, so a column with rules is still one call on the write path -- the same one call a column without rules costs today. compile_column_validator generates that fused function; the loop in validation.py never learns that constraints exist.

Three layers, in the order they run:

1. the column's PgType.coerce -- the type, unchanged and still the single source of the type rules;
2. the column's Check chain -- per-field business rules;
3. the model's Rule set -- whole-object rules over several fields.

Layers 1 and 2 are fused together and run on every write: the constructor, attribute assignment, and the request body validator. Layer 3 needs an object that is finished, so it runs where a whole object is proven at once.

Rules only ever accumulate. narrow appends checks to a column inherited from a base class and cannot remove one, so every Intern that validates is by construction a valid Employee. Widening is not expressible.

Check#

One business rule over a single value that has already passed its type.

A check never sees a value of the wrong type: coercion runs first, so Le(50_000) on an Int64 column compares two integers and can assume it. That is why a check can compile down to a bare comparison.

Subclasses implement source, which returns a Python expression that is true when the value is acceptable. Returning source rather than a predicate is what lets the whole chain fuse into one function with no call per rule.

check_type#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

source#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

CheckViolation#

A value of the right type that breaks a business rule.

This is a ValueError on purpose. Every seam that already handles a rejected assignment -- the body validator's per-field except, a constructor call, a plain obj.field = value -- handles a broken business rule the same way, with no new except clause anywhere.

CheckViolation(message: str, kind: str) -> None

Ge#

value >= bound.

source (inherited from _Comparison)#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

check_type (inherited from Check)#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

Gt#

value > bound.

source (inherited from _Comparison)#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

check_type (inherited from Check)#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

Le#

value <= bound.

source (inherited from _Comparison)#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

check_type (inherited from Check)#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

Length#

Bound the length of a string or bytes value.

Length(minimum: int | None = None, maximum: int | None = None) -> None

source#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

check_type (inherited from Check)#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

Lt#

value < bound.

source (inherited from _Comparison)#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

check_type (inherited from Check)#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

Narrow#

A subclass tightening a rule on a column it inherited.

Declared in a class body with narrow. The base's checks still run, first; these are appended. There is deliberately no way to drop an inherited check, so a narrowed model always satisfies the model it narrows.

Narrow(field: str, checks: tuple[Check, ...]) -> None

OneOf#

Restrict a column to a fixed set of values.

This is how an enum stays in Python. The database keeps enforcing the type; which of its values are meaningful is an application rule, so the schema stays portable.

OneOf(*values: Any) -> None

check_type#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

source#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

Pattern#

Require a string to contain a match for a regular expression.

The pattern is searched, not anchored: anchor it yourself with ^ and $ when that is what you mean.

Pattern(pattern: str | re.Pattern[str]) -> None

source#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

check_type (inherited from Check)#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

Predicate#

An arbitrary rule over one value: the escape hatch.

The function is called with the coerced value and returns true when it is acceptable. Prefer the specific checks where they fit -- they compile to a comparison, while this compiles to a call.

Predicate(function: Callable[[Any], bool], message: str, *, kind: str = 'predicate') -> None

source#

source(var: str, ns: dict[str, Any]) -> str

Python source for an expression that is true when var is valid.

ns is the namespace the generated function closes over; a check that needs a value it cannot write as a literal binds it there with _bind and refers to it by the returned name.

check_type (inherited from Check)#

check_type(pg_type: Any, where: str) -> None

Reject a check applied to a type it cannot mean anything for.

Rule#

A business rule spanning more than one column.

Declared in a class body with rule. Unlike a check, a rule needs an object that is finished, so it runs once the fields are all in -- not on assignment, where the other fields may not have values yet.

Rule(fields: tuple[str, ...], test: Callable[..., bool], message: str, kind: str, at: str | None) -> None

check_rules#

check_rules(instance: Any) -> list[tuple[str, str, str | None]]

Run a model's rules against a finished object.

Returns one (message, kind, at) per broken rule, empty when the object is acceptable. A rule whose columns are not all loaded is skipped rather than guessed at: a field that is absent has already been reported by whoever was proving the fields, and a rule cannot say anything useful about a value that is not there.

compile_column_validator#

compile_column_validator(column: Any, owner: str) -> Callable[[Any], Any]

Fuse a column's type and its checks into one callable.

Returns pg_type.coerce itself when the column has no checks, so a plain column keeps costing exactly one call and this whole module stays off its path. With checks, the generated function is still one call: the checks become comparisons inside it rather than calls of their own.

The result is the only thing that proves a value for this column. Both the native and the Python storage assign through it, and the body validator runs it, so the type rules and the business rules cannot drift apart -- there is one function and three callers.

compile_rules#

compile_rules(model: Any) -> tuple[CompiledRule, ...]

Resolve each rule's column names to storage indexes, once.

Names are looked up here, at class-creation time, so a rule over a column that does not exist is a declaration error rather than a surprise on the first request that happens to hit it.

narrow#

narrow(field: str, *checks: Check) -> Narrow

Add checks to an inherited column, for this model only:

class Intern(Employee, table="interns"):
    salary_cap = narrow("salary", Le(50_000))

The attribute name is documentation; the metaclass finds these by type. The base's own checks are unaffected and still run first.

rule#

rule(*fields: str, message: str | None = None, at: str | None = None, name: str | None = None) -> Callable[[Callable[..., bool]], Rule]

Declare a whole-object rule over several columns:

class Intern(Employee, table="interns"):
    @rule("salary", "tenure_months")
    def pay_band(salary: int, tenure_months: int) -> bool:
        "an intern past six months cannot be paid more than 40k"
        return not (tenure_months > 6 and salary > 40_000)

The function takes the named columns' validated values, in the order named, and returns true when the object is acceptable. It is a plain function, not a method: it receives values rather than an instance, so it never touches a descriptor and cannot read a column it did not declare.

The docstring is the error message unless message is given. By default the error is reported against the object; at="salary" reports it against one field instead, which is usually what a form wants.

wreath.orm.relations#

Relationship declarations and explicit load options.

LoadOption#

An explicit request to load one relationship with one strategy.

LoadOption(relationship: Relationship, strategy: str, nested: tuple[LoadOption, ...]) -> None

Relationship#

A declared relationship, and the descriptor that reads it.

Relationship(target: Any, *, foreign_key: Any, back_populates: str | None, load: LoadStrategy) -> None

expression (property)#

expression: RelationshipExpr

RelationshipExpr#

Class-level access to a relationship; builds load options and predicates.

RelationshipExpr(relationship: Relationship, path: tuple[Relationship, ...] = ()) -> None

joined#

joined(*nested: LoadOption) -> LoadOption

Load this relationship in the same statement with a LEFT JOIN.

selectin#

selectin(*nested: LoadOption) -> LoadOption

Load this relationship with a second batched statement.

relationship#

relationship(target: Any, *, foreign_key: Any, back_populates: str | None = None, load: LoadStrategy = 'raise') -> Any

Declare a relationship to target.

target is a model class or its name, resolved only inside the registry that owns both models. foreign_key names the column holding the key: a column on this model makes the relationship to-one, and a column on the target makes it to-many.

load defaults to "raise", so reading the attribute without loading it raises instead of emitting a query. "selectin" and "joined" load it with every query for this model.

wreath.orm.expressions#

Immutable SQL expression nodes.

Operator overloads only build nodes. Nothing here inspects a database, holds a connection, or renders SQL text; the compiler owns rendering and the session owns execution.

Nodes compare by identity, because == is overloaded to build an equality predicate rather than answer one.

BinaryExpr#

An expression that renders where a boolean is expected.

BinaryExpr(operator: str, left: Expression, right: Expression) -> None

asc#

asc() -> OrderExpr

Order by this distance or rank, smallest first.

desc#

desc() -> OrderExpr

Order by this distance or rank, largest first.

The one to reach for with .rank(): a relevance score is higher when the row is a better match, which is the opposite of a distance.

is_distance (property)#

is_distance: bool

Whether this node evaluates to a distance rather than a boolean.

is_rank (property)#

is_rank: bool

Whether this node evaluates to a text-search relevance score.

BooleanExpr#

An expression that renders where a boolean is expected.

BooleanExpr(operator: str, operands: tuple[Predicate, ...]) -> None

ColumnExpr#

A reference to one model column.

ColumnExpr(column: Any) -> None

all_eq#

all_eq(value: Any) -> BinaryExpr

value = ALL(self) -- every array element equals value.

any_eq#

any_eq(value: Any) -> BinaryExpr

value = ANY(self) -- value is an element of the array.

asc#

asc() -> OrderExpr

contained_by#

contained_by(other: Any) -> BinaryExpr

self <@ other -- reverse jsonb/array containment.

contains#

contains(other: Any) -> BinaryExpr

self @> other -- jsonb/array containment.

cosine_distance#

cosine_distance(other: Any) -> BinaryExpr

self <=> other -- cosine distance, the common default. Orderable.

covered_by#

covered_by(region: Any) -> BinaryExpr

ST_Covers(ST_GeogFromText($1), self) — rows inside a region.

The question tier 1 structurally cannot answer: a catchment, a council boundary, an exclusion zone. region is a wreath.geospatial.Polygon, which is built from Coordinates and therefore cannot arrive with its longitude and latitude the wrong way round.

ST_Covers rather than ST_Contains, and not as a preference: ST_Contains(geography, geography) does not exist, so a query reaching for it fails at the database. ST_Covers also answers "yes" for a point exactly on the boundary, which is the reading a catchment wants -- an address on the line is in the district.

The polygon travels out as text and PostGIS lifts it, so there is no bind-only column type and no OID to resolve. Storing polygons is a different feature; this one only queries against them.

Args:

• region — The Polygon to test containment in.

Returns:

• The predicate, ready for .where().

Raises:

• DeclarationError — This is not a Geography column.
• TypeError — region is not a Polygon.

desc#

desc() -> OrderExpr

hamming_distance#

hamming_distance(other: Any) -> BinaryExpr

self <~> other -- how many bits differ. Orderable.

The query half of binary quantization: over a Bit column, this counts the positions where two signatures disagree, which is the cheap stand-in for distance that lets a 32x smaller index shortlist candidates before the real vectors re-score them.

has_all#

has_all(keys: Any) -> BinaryExpr

self ?& keys -- the jsonb object has all of these keys.

has_any#

has_any(keys: Any) -> BinaryExpr

self ?| keys -- the jsonb object has any of these keys.

has_key#

has_key(key: str) -> BinaryExpr

self ? key -- the jsonb object has this top-level key.

ilike#

ilike(pattern: str) -> BinaryExpr

in_#

in_(values: Any) -> InExpr | InSubqueryExpr

self IN (...) against a list of values or a one-column Select.

Args:

• values — An iterable of values, rendered as IN ($1, $2, ...); or a
• Select projecting exactly one column, rendered as
• IN (SELECT ...) so the filter stays in the database.

Returns:

• The membership predicate, ready to pass to .where().

Raises:

• TypeError — values is neither an iterable nor a Select.
• ValueError — values is an empty list. A subquery may legitimately
• return no rows, so this applies to the list form only.
• DeclarationError — The subquery is not a shape IN accepts -- see
• _check_subquery for the four refusals and why each exists.

inner_product#

inner_product(other: Any) -> BinaryExpr

self <#> other -- negative inner product, as pgvector defines it.

Negative so that ordering ascending still puts the most similar row first, which is the whole reason pgvector spells it that way. Read a result of -0.9 as an inner product of 0.9.

is_not_null#

is_not_null() -> UnaryExpr

is_null#

is_null() -> UnaryExpr

jaccard_distance#

jaccard_distance(other: Any) -> BinaryExpr

self <%> other -- one minus the Jaccard similarity. Orderable.

Set overlap rather than positional agreement: it counts the bits set in both against the bits set in either, so two sparse signatures that share their few set bits are close even though most positions agree trivially by being zero in both. That is the difference from hamming_distance, which those shared zeros dominate.

l1_distance#

l1_distance(other: Any) -> BinaryExpr

self <+> other -- taxicab (L1) distance. Orderable.

l2_distance#

l2_distance(other: Any) -> BinaryExpr

self <-> other -- Euclidean distance. Orderable.

like#

like(pattern: str) -> BinaryExpr

matches#

matches(terms: Any, *, parser: str = 'websearch_to_tsquery') -> BinaryExpr

self @@ websearch_to_tsquery('<config>', terms) -- a text-search predicate.

The configuration comes from the column's own TsVector declaration, so the query is analysed exactly as the stored vector was; a query analysed under a different configuration matches nothing and looks like missing data.

Args:

• terms — The search text, bound as a parameter. A wreath.queries.Param
• works here as it does for the distance operators.
• parser — "websearch_to_tsquery" (the default) or "to_tsquery".
• The default never raises on user input -- ", &, ! and a
• lone — are just characters to it. to_tsquery understands
• &/|/!/<-> and raises a syntax error on malformed input,
• so choose it only where the query is the application's and the
• error is handled.

Returns:

• The predicate, ready for .where().

Raises:

• DeclarationError — This is not a TsVector column, or parser is
• not one of the two.

model (property)#

model: type

nearest#

nearest(centre: Any) -> BinaryExpr

Distance to centre in metres -- an order key, not a predicate.

order_by(Model.at.nearest(here)) with a limit is the shape an index answers. A number rather than a boolean, exactly as the pgvector distances and ts_rank are, so where() refuses it on its own.

On a Geography column this is PostGIS's <->, which the GiST index answers directly as an Index Scan ... Order By: -- a true nearest-neighbour search rather than tier 1's bounding-box approximation, and measured on the spheroid rather than on a sphere. Both answer metres, so the two tiers are comparable; they differ by the ~0.5% the guide states.

On a Point column it renders the great-circle haversine, unchanged.

not_in#

not_in(values: Any) -> InExpr | InSubqueryExpr

self NOT IN (...) against a list of values or a one-column Select.

Takes the same operands as in_, with one extra refusal: a subquery projecting a nullable column. SQL's three-valued logic makes NOT IN (SELECT ...) return no rows at all the moment the subquery yields a single NULL, because x <> NULL is unknown rather than true. That passes every test written against data without NULLs and empties a page in production, so it is refused at the call site instead.

Args:

• values — An iterable of values, or a Select projecting exactly one
• NOT NULL column.

Returns:

• The negated membership predicate.

Raises:

• DeclarationError — The subquery projects a nullable column, or any of
• the refusals in_ also makes.

overlaps#

overlaps(other: Any) -> BinaryExpr

self && other -- the arrays share at least one element.

path#

path(elements: Any, *, as_json: bool = False) -> _JsonPath

A jsonb path extraction completed by a comparison.

Model.data.path(["a", "b"]) == "x" renders (data #>> $1) = $2. as_json=True extracts a sub-document with #> instead, usable with the jsonb operators.

rank#

rank(terms: Any, *, parser: str = 'websearch_to_tsquery') -> BinaryExpr

ts_rank(self, websearch_to_tsquery('<config>', terms)) -- a relevance score.

A number, not a predicate: order by it (.rank(t).desc()) or compare it against a threshold. where() refuses one on its own, because PostgreSQL would refuse it later with a message about the argument of WHERE rather than about the line that wrote it.

Takes the same arguments as matches, and should be given the same terms and parser -- ranking one query while filtering by another is a bug that looks like bad relevance.

within#

within(centre: Any, metres: Any) -> Predicate

Rows whose point is within metres of centre, great-circle.

On a Geography column this is ST_DWithin(col, $1, $2), which PostGIS plans as an && index condition with the exact spheroidal test filtered over what survives -- literally the shape the tier-1 rendering below hand-builds, which is why both are spelled within().

On a Point column it renders as two things ANDed, and both halves are load-bearing:

• self <@ box(point(...), point(...)) -- the degree-aligned bounding box, which is the only form a GiST point_ops index can answer, and therefore the only reason this does not read the whole table.
• the exact haversine distance, compared against metres.

The box is a superset of the circle -- its corners reach about 40% further than its edges -- so dropping the second half would return rows that are not within the radius. Dropping the first half would be correct and would scan everything. Neither is optional.

A circle crossing the antimeridian produces two boxes, ORed, because <@ has no notion of a wrapped edge; one reaching a pole produces a box spanning every longitude. wreath.geospatial.bounding_boxes makes that decision and this method only renders it.

Expression#

Base class for everything that renders into a SQL fragment.

InExpr#

Membership against an explicit value list.

Rendered as IN ($1, $2, ...) because the driver codecs exchange scalars rather than arrays; the operand count is therefore part of the query shape.

InExpr(operator: str, left: Expression, values: tuple[ValueExpr, ...]) -> None

InSubqueryExpr#

Membership against a one-column subquery: left IN (SELECT ...).

Distinct from InExpr rather than a mode of it, because the two differ in every respect the compiler cares about: the operand count is not part of the query shape (the subquery's own shape is), the bound values live inside a nested Select rather than beside the operator, and the rendered text is a statement rather than a parenthesised list. Sharing one class would mean a branch at each of those points and an attribute that is meaningless in one of the two states.

select is validated at construction by _check_subquery, so a node that exists is one the compiler can render.

InSubqueryExpr(operator: str, left: Expression, select: Any) -> None

OrderExpr#

One ORDER BY key: a column, or an expression that evaluates to one.

expression is a bare ColumnExpr for an ordinary sort, and a BinaryExpr for the two kinds of ordering that are computed rather than stored -- a vector distance and a text-search rank. The compiler branches on which, because a computed key carries a bound value and so has to render through the operand renderer rather than as a quoted column name.

OrderExpr(expression: Expression, direction: str) -> None

Predicate#

An expression that renders where a boolean is expected.

UnaryExpr#

An expression that renders where a boolean is expected.

UnaryExpr(operator: str, operand: Expression) -> None

ValueExpr#

One bound parameter. The value never reaches SQL text or a cache key.

ValueExpr(value: Any, pg_type: Any) -> None

and_#

and_(*predicates: Predicate) -> Predicate

Combine predicates with AND; a single predicate passes through.

not_#

not_(predicate: Predicate) -> UnaryExpr

or_#

or_(*predicates: Predicate) -> Predicate

Combine predicates with OR; a single predicate passes through.

wreath.orm.query#

Immutable query objects.

Every builder method returns a new Select; nothing mutates in place, so a query can be built once at import time and reused per request without a defensive copy.

Select#

A compiled-on-demand SELECT for one model.

Select(model: type, projection: tuple[ColumnExpr, ...], predicates: tuple[Predicate, ...], includes: tuple[LoadOption, ...], orderings: tuple[OrderExpr, ...], limit_: int | None, offset_: int | None, for_update_: bool) -> None

build (classmethod)#

build(model: type, fields: tuple[Any, ...]) -> Select

for_update#

for_update() -> Select

Lock matched rows; requires a write session inside a transaction.

include#

include(*load_options: LoadOption) -> Select

Load relationships with this query.

limit#

limit(value: int) -> Select

offset#

offset(value: int) -> Select

order_by#

order_by(*expressions: Any) -> Select

Order rows by columns, .asc()/.desc(), or a vector distance.

A distance orders by similarity, which is the shape a vector search actually has:

Document.select().order_by(Document.embedding.cosine_distance(query))

Ordering by a distance is what an HNSW or IVFFlat index answers; a distance in a WHERE is not, and the index will not be used for it.

paginate#

paginate(page: int, size: int) -> Select

Shape this query for one page: LIMIT size OFFSET (page - 1) * size.

Pure query-shaping; execution and the total count live in wreath.pagination.

rebound#

rebound(predicates: tuple[Predicate, ...]) -> Select

This query with its predicates replaced rather than extended.

Every other builder method adds to a query; this one swaps one set of predicates for another, which is what a declared query in wreath.queries needs — it fixes a shape once and substitutes each call's values into it. The caller owes the invariant that the replacements have the same tree structure and column types as the originals; anything else changes the plan-cache key, and a declared query compiling once per shape is the whole point of declaring it.

rebound_orderings#

rebound_orderings(orderings: tuple[OrderExpr, ...]) -> Select

This query with its ORDER BY keys replaced rather than extended.

The ordering counterpart of rebound, and it exists for the same caller: a declared vector search fixes ORDER BY embedding <=> $n at class-definition time and substitutes the query vector per call. Same invariant -- the replacements must have the same tree structure and column types, because anything else is a different plan-cache key.

where#

where(*predicates: Predicate) -> Select

Add predicates; repeated calls and multiple arguments combine with AND.

wreath.orm.session#

Request-scoped sessions.

A session owns one leased connection, one identity map, and the pending write sets. It acquires the connection on first use and returns it exactly once, on success, error, or cancellation.

Nothing here loads data implicitly: every statement this module issues is the direct result of an awaited call.

FromORM#

Marks a handler parameter as a request-scoped ORM session.

database names an app.orm() registry; it may be omitted when the application has exactly one.

tenant says where the tenant binding comes from for an isolated registry, and wreath.tenancy.FromTenant() is the usual answer -- the tenant this request resolved to. It exists because without it the only route to a tenant-bound session was constructing Session(registry, workload, tenant=...) by hand in the handler body, which made the framework's own convenience the unsafe spelling and hand-rolling the safe one. An isolated registry with no tenant here is refused at route-compile time.

FromORM(database: str | None = None, workload: Workload = 'read', tenant: Any = None) -> None

RawQuery#

Unmodified SQL, executed on this session's leased connection.

Wreath does not parse, rewrite, or cache the SQL; results come back as the driver's own Record objects unless models() is used.

RawQuery(session: Session, sql: str, args: tuple[Any, ...]) -> None

execute#

async execute() -> str

fetch#

async fetch() -> list[Any]

fetchrow#

async fetchrow() -> Any

fetchval#

async fetchval() -> Any

models#

async models(model: type) -> list[Any]

Hydrate this result into model instances.

The result must contain every column of the model exactly once, named as in the database. Extra or missing columns are rejected rather than silently dropped, so a drifting query fails loudly.

Session#

One unit of work over one leased connection.

The identity map holds every object this session hydrated, for as long as the session is open. That is what makes the map an identity map -- two reads of one row give one object, and a pending change is not lost to a refetch -- and it means a fetch of a million rows pins a million objects until close(). There is deliberately no ceiling: evicting an entry would detach an object the caller may still be holding, which changes what the ORM means rather than bounding it. Past identity_map_warn_at the session emits one ResourceWarning, so the situation is visible before it is a memory incident. Page large reads, or open a session per page.

Session(registry: Any, workload: Workload, *, tenant: TenantContext | None = None, statement_timeout: float | None = None, identity_map_warn_at: int | None = None, audit: Any = None) -> None

add#

add(instance: Any) -> None

Schedule instance for insertion on the next flush.

begin#

begin() -> Any

Open a transaction, or a savepoint when one is already open.

check_identity_map#

check_identity_map() -> None

Warn, once, if this session is holding an unusual number of objects.

Nothing calls this for you. Every automatic placement costs a boundary crossing on a path the whole application pays for -- per fetch, or per session close -- and wreath-request-trace prices each at +1 on the realistic scenario. That is a poor trade for a diagnostic that fires for almost nobody, so the check is a thing you call while chasing a memory problem:

rows = await session.fetch(Big.select()) session.check_identity_map()

Set identity_map_warn_at on the session or the registry to pick the threshold; without one this does nothing.

close#

async close() -> None

Roll back any open transaction and return the connection exactly once.

closed (property)#

closed: bool

connection (property)#

connection: Any

The leased connection, or None while the session is still lazy.

count#

async count(query: Select) -> int

Count the rows query matches, without fetching or hydrating them.

Emits SELECT COUNT(*) with the query's filters (and any to-one filter joins), ignoring its projection, ordering, and paging. This is what wreath.pagination.paginate uses for the page total, so a large result set costs one aggregate round trip rather than transferring and materializing every matching row.

create#

async create(model: type, **values: Any) -> Any

Construct, insert, and return one mapped model.

The model constructor remains the single validation path. This is a session convenience, not a second model or repository abstraction.

declared#

async declared(sql: str, args: tuple[Any, ...]) -> list[Any]

Run one statement rendered from a declaration, and return its rows.

The seam wreath.series executes through. It exists rather than going via raw because a calculated view is compiled from model metadata the same way a Select is, so it owes the same two checks a Select gets: that the session is open, and that a tenant session is inside the transaction its search_path is bound by. Statement text comes from the compiler, never from a caller.

delete#

delete(instance: Any) -> None

Schedule instance for deletion on the next flush.

delete_where#

async delete_where(query: Select) -> int

Delete rows selected by an explicit predicate and return its count.

fetch#

async fetch(query: Select) -> list[Any]

Run query and hydrate every row.

fetch_one#

async fetch_one(query: Select) -> Any

Run query and return one object, or None.

Raises MultipleResultsError when the query matches more than one row; a stricter limit set by the caller is left alone.

flush#

async flush() -> None

Write pending inserts, updates, and deletes in a deterministic order.

Outside an explicit transaction this opens one for the flush and commits or rolls it back atomically.

That is also what makes this safe for an isolated-tenant session without the _check_tenant_bound guard the read paths carry: the begin() below binds search_path before any statement runs, so there is no window where tenant-template SQL sees the pooled connection's previous namespace. It is load-bearing rather than incidental -- do not "optimise" the transaction away for a single- statement flush.

get#

async get(model: type, primary_key: Any, *, load: Iterable[Any] = ()) -> Any

Fetch one object by primary key, or None.

in_transaction (property)#

in_transaction: bool

load#

async load(target: Any, relationship: Any) -> None

Load relationship on one object or a sequence, in one batch.

lock#

async lock(key: str, *, scope: str = 'xact', mode: str = 'exclusive', namespace: str | None = None) -> None

Take a transaction-scoped PostgreSQL advisory lock on this session.

scope="xact" (the default and recommended form) takes a lock that PostgreSQL releases automatically at COMMIT/ROLLBACK; it must run inside async with session.begin(): and rides the connection the transaction already pins, so there is no explicit unlock and no connection-affinity bookkeeping.

Advisory locks ignore search_path, so for an isolated-tenant session the tenant schema is folded into the lock namespace automatically -- two tenants never collide on the same key. Pass namespace to override (for example, to take a deliberately fleet-global lock).

Session-scoped locks (held beyond the transaction) are intentionally not offered here: releasing them correctly is bound to returning the pooled connection, which database.lock(...) / database.try_lock(...) own.

raw#

raw(sql: str | Template, *args: Any) -> RawQuery

Run SQL exactly as written on this session's connection.

A t-string is the spelling to reach for when any part of the statement is a value. wreath.sql compiles it to $1-style placeholders and binds what was interpolated, so the value cannot become syntax:

await db.raw(t"SELECT id FROM shipments WHERE reference = {ref}").fetch()

A plain str is still accepted, and still means what it always meant: SQL wreath does not parse, rewrite, or cache, with $1 placeholders bound from args. That is the right tool for a statement written in full at the call site. It is the wrong one for a statement assembled from a caller's text, which is what wreath.hardening's SQL101 exists to say -- one character at the quote turns the wrong one into the right one.

registry (property)#

registry: Any

require#

async require(model: type, primary_key: Any, *, load: Iterable[Any] = ()) -> Any

Fetch one object by primary key, raising when it does not exist.

require_one#

async require_one(query: Select) -> Any

Run query and require exactly one result.

update_where#

async update_where(query: Select, **values: Any) -> int

Update rows selected by an explicit predicate and return its count.

workload (property)#

workload: Workload

TenantContext#

A validated, request-scoped tenant binding for an isolated registry.

schema is the tenant's physical PostgreSQL schema; role is an optional database role selected transaction-locally when the deployment relies on PostgreSQL to enforce hostile-tenant isolation. Both are validated as unquoted identifiers at construction, so the transaction-local SET LOCAL statements the session issues can never carry untrusted text.

A context must be resolved from an application-owned tenant directory, never from a request host, path, header, or token. It is transaction-local by construction: the session binds it after BEGIN and PostgreSQL discards it at transaction end, so a pooled connection never leaks one tenant's binding into the next lease.

TenantContext(schema: str, role: str | None = None) -> None

compile_session_binding#

compile_session_binding(registries: Any, marker: FromORM) -> tuple[str, Any]

Resolve marker to a registry at route-compile time.

wreath.orm.table#

Table-level schema declarations found in a model body by type.

These sit alongside narrow and rule: you declare them as plain class attributes and the metaclass collects them by type, so the attribute name is only documentation:

class Membership(Model, table="memberships", schema=TENANT_SCHEMA):
    org_id: Mapped[int] = column(Int64)
    user_id: Mapped[int] = column(Int64)
    _identity = unique("org_id", "user_id")
    _lookup = index("user_id", "org_id")

Constraints and indexes declared this way are named deterministically by the migration engine (never by you), so a downgrade drops exactly what an upgrade created. Columns are named by their database name (which is the attribute name).

AllOf#

Two or more predicates joined by AND.

AllOf(operands: tuple[object, ...]) -> None

Eq#

column = literal.

Eq(column: str, value: object) -> None

Facet#

A declaration another subsystem attaches to a model, found by type.

Columns are the ORM's. What an audit trail wants to redact, what a privacy traversal considers personal, what a sync shape keys on -- those are facts about a model that the ORM does not own and must not have to know about. A facet is how a subsystem outside the ORM declares one on the model itself, next to the columns it talks about, rather than in a registry that drifts from the schema on the next migration.

class Photo(Model, table="photos"):
    owner_id: Mapped[UUID] = column(Uuid)
    exif_gps: Mapped[str | None] = column(Text, null=True)

    _audit = audited(redact={"exif_gps"})

Subclass it, give it a namespace, and list in columns every column name the facet mentions. The metaclass does two things with that, and both are the things a subsystem would otherwise get wrong on its own:

• it validates the column names when the class is created, so a facet naming a column the model does not declare is a DeclarationError pointing at the class body, not a KeyError on the first write;
• it refuses two facets sharing a namespace on one model, so "which of the two audit declarations wins" is never a question.

One mechanism rather than one per subsystem: audited(...) and a privacy classification are the same shape with different payloads, and the class kwarg each would otherwise need (Model(audited=True, redact=...)) does not compose -- two subsystems eventually want the same keyword, and the metaclass grows a parameter per feature.

Attribute names are documentation, exactly as for unique and index: the metaclass collects by type.

Facet(columns: tuple[str, ...] = ()) -> None

Index#

A multi-column btree index, optionally UNIQUE and optionally partial.

where carries a predicate built from eq, is_null, is_not_null, one_of and all_of. Columns are named by string for the same reason index names them by string -- a model cannot refer to its own attributes from inside its own class body, so the query language (Job.state == "ready") is not available here.

where_sql is the rendered predicate in PostgreSQL's own normal form, set by the registry once column types are known. It is None until then.

Index(columns: tuple[str, ...], unique: bool, where: object = None, where_sql: str | None = None) -> None

IsNull#

column IS NULL, or IS NOT NULL when negated.

IsNull(column: str, negated: bool) -> None

InValues#

column IN (...) over two or more literals.

InValues(column: str, values: tuple[object, ...]) -> None

Unique#

A composite UNIQUE constraint over two or more columns.

Unique(columns: tuple[str, ...]) -> None

all_of#

all_of(*predicates: object) -> AllOf

Two or more predicates joined by AND.

eq#

eq(column: str, value: object) -> Eq

column = value, for a text, integer, or boolean column.

facet#

facet(model: object, namespace: str) -> Facet | None

The declaration namespace attached to model, or None.

The read side of Facet. None rather than a raise, because "this model is not audited" is the ordinary answer for most models and making every caller guard would put a try around every read.

index#

index(*columns: str, unique: bool = False, where: object = None) -> Index

Declare a multi-column btree index by column name.

unique=True makes it a unique index; where= makes it partial:

_claim = index("queue", "run_at", where=eq("state", "ready"))
_dedup = index("queue", "dedup_key", unique=True, where=is_not_null("dedup_key"))

is_not_null#

is_not_null(column: str) -> IsNull

column IS NOT NULL -- the shape a unique partial index usually wants.

is_null#

is_null(column: str) -> IsNull

column IS NULL.

one_of#

one_of(column: str, values: object) -> InValues

column IN (...), over two or more literals of one type.

A single-element list is refused: PostgreSQL rewrites IN ('a') to = 'a', so the catalog would never match what was declared. Use eq, which is what that means anyway.

unique#

unique(*columns: str) -> Unique

Declare a table-level composite unique constraint by column name.

wreath.orm.types#

Explicit PostgreSQL column types.

Every type names one PostgreSQL OID that both driver backends encode and decode. There is no inference and no implicit widening: a column declares the type it has in the database, and values that do not fit it are rejected before any SQL runs.

coerce validates and normalizes a Python value on assignment. to_wire and from_wire convert between that Python value and the representation the driver codec exchanges, and are the identity for everything except JSON.

Geography#

Geography(*, srid: int = 4326) -> ExtensionType

Declare a PostGIS geography(Point,srid) column.

The Python value is a wreath.geospatial.Coordinate, the same value type a tier-1 Point column holds, so a model can move between the tiers without its handlers changing.

class Station(Model, table="stations"):
    at: Mapped[Coordinate] = column(Geography(), index="gist")

Only the point form is declarable. geography also spells polygons and line strings, and wreath has no value type for either -- a geography(Polygon,4326) column would have nothing to hold and nothing to validate, so it is left to a hand-written migration rather than declared with a hole in it.

Requires CREATE EXTENSION postgis in the column's database. A registry that declares one against a database without it fails at startup, naming the extension and the schema, rather than at the first query with an OID error -- the same contract Vector has, because it is the same mechanism.

Tier 1 needs none of this. Point stores a coordinate in core PostgreSQL and answers "within" and "nearest" through a core GiST index. Reach for Geography when you need what tier 1 does not do: a true KNN ordering, containment against a polygon, or a projection.

Halfvec#

Halfvec(dim: int) -> ExtensionType

Declare a pgvector halfvec(dim) column -- half-precision, half the storage.

The same Python value as Vector: a list[float] of exactly dim finite floats. What differs is what the database keeps. Each element is stored as an IEEE-754 binary16 rather than a binary32, which halves both the column and any HNSW index over it -- the reason to reach for this at 1536 dimensions and up, where the index is usually the thing that stopped fitting in memory.

class Document(Model, table="documents"):
    embedding: Mapped[list[float]] = column(
        Halfvec(1536), index="hnsw", index_ops="halfvec_cosine_ops"
    )

Be deliberate about the precision. binary16 carries about three decimal digits, so a value written and read back is not the value you wrote: 0.1 returns 0.0999755859375. For embedding similarity that is normally irrelevant -- the ranking is what matters and it is robust to the third digit -- but a halfvec is the wrong type for anything you intend to compare for equality or accumulate.

Two refusals, both at coercion time rather than at the server:

• A magnitude above MAX_HALF_MAGNITUDE (65504) would round to an infinity, which pgvector rejects. Refused here so the error names the element rather than arriving as an INSERT failure naming neither it nor the column.
• NaN and infinity, exactly as for Vector.

Note the operator classes are the type's own -- halfvec_cosine_ops, not vector_cosine_ops. Naming vector's opclass on a halfvec column is an error pgvector reports at index creation.

Requires CREATE EXTENSION vector (the same extension provides both types).

Array#

Array(element: PgType, *, nullable_elements: bool = False) -> _ArrayType

Declare a one-dimensional PostgreSQL array of element.

Each element is validated and wired through element's own rules, so Array(Uuid) accepts a list/tuple of UUIDs and rejects anything else. Elements are non-nullable unless nullable_elements=True. Nested arrays are not supported: an element type may not itself be an array.

Bit#

Bit(length: int) -> PgType

Declare a PostgreSQL bit(length) column -- one bit per dimension.

This is the storage half of binary quantization: an embedding whose elements are replaced by their signs, so 1,536 float4s (6,148 bytes) become 1,536 bits (192 bytes), a 32x reduction that keeps enough of the geometry to rank candidates before a re-scoring pass over the real vectors.

class Document(Model, table="documents"):
    signature: Mapped[str] = column(
        Bit(1536), index="hnsw", index_ops="bit_hamming_ops"
    )

The Python value is a str of '0' and '1', exactly length of them -- what psql shows and what '101'::bit(3) means. bytes is accepted on the way in as a convenience for the quantizers that produce it (numpy.packbits(...).tobytes()), and is unpacked using the declared length; the trailing bits of the final byte must be zero, since they name positions the column does not have. A read always returns the str, because a bytes return would have to carry its own bit count to be unambiguous.

bit is a built-in type; the operators are pgvector's. hamming_distance and jaccard_distance (<~> and <%>) and the bit_hamming_ops / bit_jaccard_ops index classes come from CREATE EXTENSION vector. So a Bit column needs no extension to store, and there is no OID to resolve -- but a query that ranks by one of those distances does.

Bool (value)#

Bool: PgType = <PgType bool oid=16>

Bytea (value)#

Bytea: PgType = <PgType bytea oid=17>

Date (value)#

Date: PgType = <PgType date oid=1082>

ExtensionType#

A type an extension adds, whose OID the database assigns.

Every built-in type here names a compile-time OID. An extension type cannot: CREATE EXTENSION vector allocates vector's OID from the same sequence as everything else in that database, so it differs between one database and the next, and between a schema and the one beside it if the extension was installed twice.

Three consequences shape this class.

• oid starts at 0 and is filled in at startup, by wreath.orm.introspection.resolve_extension_types, which reads pg_catalog once and hands the answer to both driver backends. Binding a value before that raises rather than encoding against OID 0.
• shape_value is derived from the type's name, never its OID. It is the type's contribution to the plan-cache key: an OID in there would give the same query two cache entries against two databases, and would change under a reinstall of the extension. The failure mode is a silently duplicated plan cache, which is invisible until someone profiles it.
• fingerprint_oid is likewise name-derived, for the same reason one layer up: a model fingerprint that moved with the database would report drift that is not there.

Nothing here is specific to pgvector. hstore, citext, and geometry would all resolve through the same path.

ExtensionType(extension: str, type_name: str, sql: str, coerce: Callable[[Any], Any], *, kind: int, to_wire: Callable[[Any], Any] | None = None, from_wire: Callable[[Any], Any] | None = None) -> None

require_oid#

require_oid(doing: str) -> int

This type's resolved OID, refusing if resolution never reached it.

Call this wherever the OID decides something -- wire framing, a migration descriptor -- and never where an unresolved type is the legitimate state (declaration, fingerprinting, introspection, the probe that is about to resolve it).

The unbound state is not only "startup has not run yet". bind_extension_oid walks the types declared when it is called, so an ExtensionType constructed afterwards -- a model class defined after the application started -- keeps OID 0 until something resolves again. Zero is a legal-looking OID that means "unspecified" on the wire and "built-in" in a descriptor, so it has to be refused rather than framed.

to_wire#

to_wire(value: Any) -> Any

coerce (inherited from PgType)#

coerce(value: Any) -> Any

Validate value for this type, returning the normalized value.

from_wire (inherited from PgType)#

from_wire(value: Any) -> Any

Float4 (value)#

Float4: PgType = <PgType float4 oid=700>

Float8 (value)#

Float8: PgType = <PgType float8 oid=701>

GeneratedType#

A type whose column value the database computes, never the application.

A column declared with one of these renders as GENERATED ALWAYS AS (<expression>) STORED, so PostgreSQL recomputes it inside the same statement that changed the columns it reads. That is what keeps a derived index correct without a trigger, and it is also why such a column is not writable: an INSERT or UPDATE naming one is an error, so coerce refuses the value at the assignment rather than at the flush.

The expression itself is not held here. It names other columns, whose database names and types are only known once the registry has compiled the model, so it is rendered there -- see wreath.orm._generated.

coerce (inherited from PgType)#

coerce(value: Any) -> Any

Validate value for this type, returning the normalized value.

from_wire (inherited from PgType)#

from_wire(value: Any) -> Any

to_wire (inherited from PgType)#

to_wire(value: Any) -> Any

Float32 (value)#

Float32: PgType = <PgType float4 oid=700>

Float64 (value)#

Float64: PgType = <PgType float8 oid=701>

Int2 (value)#

Int2: PgType = <PgType int2 oid=21>

Int4 (value)#

Int4: PgType = <PgType int4 oid=23>

Int8 (value)#

Int8: PgType = <PgType int8 oid=20>

Int16 (value)#

Int16: PgType = <PgType int2 oid=21>

Int32 (value)#

Int32: PgType = <PgType int4 oid=23>

Int64 (value)#

Int64: PgType = <PgType int8 oid=20>

Json (value)#

Json: PgType = <PgType json oid=114>

Jsonb (value)#

Jsonb: PgType = <PgType jsonb oid=3802>

Numeric (value)#

Numeric: PgType = <PgType numeric oid=1700>

PgType#

One PostgreSQL type, its OID, and its Python conversion rules.

PgType(name: str, oid: int, sql: str, coerce: Callable[[Any], Any], *, to_wire: Callable[[Any], Any] | None = None, from_wire: Callable[[Any], Any] | None = None) -> None

coerce#

coerce(value: Any) -> Any

Validate value for this type, returning the normalized value.

from_wire#

from_wire(value: Any) -> Any

to_wire#

to_wire(value: Any) -> Any

Point (value)#

Point: PgType = <PgType point oid=600>

SparseVector#

A pgvector sparsevec value: a dimension, and the positions that are not zero.

Construct one from a mapping of 1-based index to value, or from anything dict() accepts (an iterable of pairs, another SparseVector's to_dict()):

SparseVector(5, {1: 1.5, 3: 3.5})       # '{1:1.5,3:3.5}/5' to pgvector
SparseVector.from_dense([1.5, 0, 3.5, 0, 0])   # the same value

Explicit zeros are dropped rather than stored. That is what the server does to them, so keeping them would mean a value did not survive its own round trip -- write {2: 0.0}, read back {} -- and a difference that appears only after a write is the worst kind to debug.

Immutable in practice: indices and values are tuples and there is no setter. Two values are equal when their dimension and their non-zero elements are, so SparseVector(5, {1: 1.0}) equals itself across a round trip through the database.

Attributes: dim: The declared dimension -- how many positions exist, not how many are stored. indices: The stored positions, 1-based and ascending. values: The stored values, positionally paired with indices.

SparseVector(dim: Any, elements: Any = ()) -> None

dim (property)#

dim: int

The declared dimension.

from_dense (classmethod)#

from_dense(values: Any) -> SparseVector

The sparse form of a dense sequence, dropping its zeros.

The dimension is the sequence's length, so a 30,000-word bag whose vocabulary is fixed keeps its dimension even when a document uses nine of the words.

indices (property)#

indices: tuple[int, ...]

The stored 1-based positions, materialized at the Python boundary.

to_dict#

to_dict() -> dict[int, float]

The non-zero elements, 1-based index to value.

values (property)#

values: tuple[float, ...]

The stored values, materialized at the Python boundary.

Sparsevec#

Sparsevec(dim: int) -> ExtensionType

Declare a pgvector sparsevec(dim) column -- only the non-zero positions.

The Python value is a SparseVector, not a list. That is the whole difference from Vector: a dimension here can be a million, and storing a million floats to say that nine of them are non-zero is what the type exists to avoid. A dict alone would not do, because it names no dimension, and the dimension is not recoverable from the elements.

from wreath.postgres import SparseVector

class Document(Model, table="documents"):
    terms: Mapped[SparseVector] = column(
        Sparsevec(30000), index="hnsw", index_ops="sparsevec_l2_ops"
    )

await session.add(Document(terms=SparseVector(30000, {17: 0.9, 4021: 1.4})))

Indices are 1-based, matching pgvector's own text form ('{1:1.5,3:3.5}/5'); the binary wire format is 0-based and the conversion lives in the codec, so the two numberings never both appear in application code.

The dimension is checked at coercion: a SparseVector declaring a different one is refused here rather than by the server, because the server's message names neither the column nor which of the two dimensions it expected.

At most MAX_SPARSEVEC_NNZ (16,000) elements may be non-zero, whatever the dimension -- pgvector's own ceiling, and a value denser than that wants Vector or Halfvec instead.

Requires CREATE EXTENSION vector (the same extension provides all three types). Note the operator classes are the type's own -- sparsevec_l2_ops, not vector_l2_ops -- and that pgvector indexes a sparsevec to 1,000 non-zero elements even though the column may hold 16,000.

Text (value)#

Text: PgType = <PgType text oid=25>

TextArray (value)#

TextArray: _ArrayType = <PgType text[] oid=1009>

Timestamp (value)#

Timestamp: PgType = <PgType timestamp oid=1114>

TimestampTz (value)#

TimestampTz: PgType = <PgType timestamptz oid=1184>

TsVector#

TsVector(config: str = 'english', *, sources: Any) -> TsVectorType

Declare a generated tsvector column over sources.

class Document(Model, table="documents"):
    title: Mapped[str] = column(Text)
    body: Mapped[str] = column(Text)
    search: Mapped[bytes] = column(
        TsVector("english", sources=("title", "body")), index="gin"
    )

That renders GENERATED ALWAYS AS (to_tsvector('english', coalesce(title,'') || ' ' || coalesce(body,''))) STORED, which is the form that keeps the GIN index correct without a trigger: PostgreSQL recomputes the column inside the same statement that changed title or body, so there is no window where the index disagrees with the row.

Each source must be a declared Text column of the same model. Other types are refused because PostgreSQL's own deparsing of the expression casts them ((COALESCE(vch, ''::character varying))::text), and a spelling wreath cannot predict is a permanent, self-renewing migration drift -- the same reason wreath.orm._index_predicate keeps its vocabulary small.

The value is the database's, not the application's: assigning one raises, and passing one to the constructor raises. Query it with .matches() and .rank() rather than reading it.

TsVectorType#

A tsvector PostgreSQL derives from other columns of the same row.

Carries the two things the DDL needs and nothing else: the text-search config to analyse under, and the sources to analyse. The expression is rendered once the registry knows those columns' database names and types.

TsVectorType(config: str, sources: tuple[str, ...]) -> None

coerce (inherited from PgType)#

coerce(value: Any) -> Any

Validate value for this type, returning the normalized value.

from_wire (inherited from PgType)#

from_wire(value: Any) -> Any

to_wire (inherited from PgType)#

to_wire(value: Any) -> Any

Uuid (value)#

Uuid: PgType = <PgType uuid oid=2950>

Varchar (value)#

Varchar: PgType = <PgType varchar oid=1043>

Vector#

Vector(dim: int) -> ExtensionType

Declare a pgvector vector(dim) column.

The Python value is a list[float] of exactly dim finite floats. Coercion is strict in both directions that matter: a wrong length is a dimension mismatch the database would reject anyway, and NaN or infinity is refused because pgvector stores neither and every distance involving one is meaningless.

class Document(Model, table="documents"):
    embedding: Mapped[list[float]] = column(
        Vector(1536), index="hnsw", index_ops="vector_cosine_ops"
    )

Requires CREATE EXTENSION vector in the column's database. A registry that declares one against a database without it fails at startup, naming the extension and the schema, rather than at the first query with an OID error.

bind_extension_oid#

bind_extension_oid(type_name: str, oid: int) -> int

Give every declared type named type_name the OID this database uses.

Returns how many types were bound. Idempotent for a repeated identical OID, which is the ordinary case when several registries share one database.

Raises:

• ValueError — the same type resolved to a different OID than a previous
• call. One process holds one codec table, so two databases whose
• vector OIDs disagree cannot both be served from it; that is
• reported here rather than by decoding one database's rows with the
• other's rules.

declared_extension_types#

declared_extension_types() -> tuple[ExtensionType, ...]

Every extension type declared in this process, in declaration order.

wreath.orm.dto#

Dataclass projections over mapped models.

Request and response models often repeat a subset of an ORM model's columns. model_dataclass makes that subset explicit while keeping the result an ordinary dataclass understood by Wreath's binding, OpenAPI, and typegen layers:

    LlamaCreate = model_dataclass(
        Llama,
        include={"name", "birth_date"},
        name="LlamaCreate",
    )

The projection is compiled when the declaration executes, never while binding a request. Its cache belongs to the mapped model itself, so two declarations of the same shape return the same type without a process-global registry.

model_dataclass#

model_dataclass(model: type[Model], *, include: Iterable[str] | None = None, exclude: Iterable[str] = (), name: str | None = None) -> type

Compile an ordinary dataclass from selected mapped columns.

include and exclude are mutually exclusive. Unknown names are refused rather than ignored, because a stale projection silently omitting a renamed field changes the wire schema. Fields retain model order and Python defaults; nullable, generated, primary-key, and server-default columns may be omitted and default to None. All fields are keyword-only, matching Wreath's request binder and avoiding declaration-order restrictions.

wreath.orm.registry#

The application-owned model registry.

A registry binds a set of models to one database. It is the only place that resolves relationships, and the only owner of compiled query plans. Two applications can register the same model classes against different databases without sharing sessions, identity objects, or cached plans.

Registry#

Compiled models for one database.

Registry(database: Any, models: Any, *, validate_schema: ValidateSchema = 'error', query_cache_size: int = 512, query_cache_bytes: int = 8388608, schema_mode: SchemaMode | None = None, statement_timeout: float | None = None, identity_map_warn_at: int | None = None) -> None

cached_plan#

cached_plan(shape_key: bytes) -> Any

cached_plan_count (property)#

cached_plan_count: int

cached_prepared_plan#

cached_prepared_plan(declaration: Any) -> tuple[bytes, Any] | None

The live plan previously associated with declaration, if any.

compile#

compile(models: tuple[type[Model], ...]) -> None

Resolve every model and relationship, then freeze the metadata.

order_of#

order_of(model: type[Model]) -> int

The model's position in specs: its dependency order for writes.

remember_prepared_shape#

remember_prepared_shape(declaration: Any, shape_key: bytes) -> None

Associate an explicit declaration with its registry-owned plan key.

spec_for#

spec_for(model: type[Model]) -> ModelSpec

store_plan#

store_plan(shape_key: bytes, plan: Any) -> Any

wreath.orm.schema#

Frozen metadata produced by registry compilation.

These objects are the only model description the request path reads. They are immutable: once Registry.compile() returns, nothing re-derives them.

CENTRAL_SCHEMA (value)#

CENTRAL_SCHEMA: SchemaRef = SchemaRef(kind='central', name=None)

TENANT_SCHEMA (value)#

TENANT_SCHEMA: SchemaRef = SchemaRef(kind='tenant', name=None)

ColumnRef#

A resolved foreign-key target, with its referential actions.

on_delete/on_update are single PostgreSQL confdeltype-style codes (a no action, r restrict, c cascade, n set null, d set default) so the desired signature compares byte-for-byte with the catalog.

ColumnRef(schema: str | SchemaRef, table: str, column: str, position: int, model_type: type[Model], on_delete: str = 'a', on_update: str = 'a', deferrable: bool = False) -> None

ColumnSpec#

ColumnSpec(python_name: 'str', database_name: 'str', position: 'int', pg_type: 'PgType', nullable: 'bool', primary_key: 'bool', unique: 'bool', indexed: 'bool', default: 'Any', server_default: 'str | None', reference: 'ColumnRef | None', column: 'Column', generated_sql: 'str | None' = None)

ColumnSpec(python_name: str, database_name: str, position: int, pg_type: PgType, nullable: bool, primary_key: bool, unique: bool, indexed: bool, default: Any, server_default: str | None, reference: ColumnRef | None, column: Column, generated_sql: str | None = None) -> None

index (property)#

index: int

index_method (property)#

index_method: str | None

The access method for this column's index, if any.

index_ops (property)#

index_ops: str | None

The operator class this column's index uses, if it names one.

index_with (property)#

index_with: tuple[tuple[str, str], ...]

This column's index-method options, as ordered (name, value) pairs.

oid (property)#

oid: int

This column's PostgreSQL type OID.

Read through the type rather than copied at compile time, because an extension type's OID is not known until startup resolves it against the live catalog -- a snapshot taken here would be 0 forever. Every built-in type answers the same constant it always did.

ModelSpec#

ModelSpec(model_type: 'type[Model]', schema: 'str', schema_ref: 'SchemaRef', table: 'str', sql_namespace: "Literal['qualified', 'tenant_search_path']", columns: 'tuple[ColumnSpec, ...]', primary_key: 'tuple[ColumnSpec, ...]', relationships: 'tuple[RelationshipSpec, ...]', storage: 'StorageSpec', by_name: 'dict[str, ColumnSpec]', by_database_name: 'dict[str, ColumnSpec]', by_relationship_name: 'dict[str, RelationshipSpec]' = <factory>, table_uniques: 'tuple[Any, ...]' = (), table_indexes: 'tuple[Any, ...]' = (), fingerprint: 'bytes' = b'')

ModelSpec(model_type: type[Model], schema: str, schema_ref: SchemaRef, table: str, sql_namespace: "Literal['qualified', 'tenant_search_path']", columns: tuple[ColumnSpec, ...], primary_key: tuple[ColumnSpec, ...], relationships: tuple[RelationshipSpec, ...], storage: StorageSpec, by_name: dict[str, ColumnSpec], by_database_name: dict[str, ColumnSpec], by_relationship_name: dict[str, RelationshipSpec] = <factory>, table_uniques: tuple[Any, ...] = (), table_indexes: tuple[Any, ...] = (), fingerprint: bytes = b'') -> None

qualified_name (property)#

qualified_name: str

relationship#

relationship(name: str) -> RelationshipSpec | None

template_name (property)#

template_name: str

RelationshipSpec#

RelationshipSpec(name: 'str', target: 'ModelSpec', local_columns: 'tuple[ColumnSpec, ...]', remote_columns: 'tuple[ColumnSpec, ...]', cardinality: "Literal['one', 'many']", default_load: "Literal['raise', 'selectin', 'joined']", relationship: 'Any')

RelationshipSpec(name: str, target: ModelSpec, local_columns: tuple[ColumnSpec, ...], remote_columns: tuple[ColumnSpec, ...], cardinality: "Literal['one', 'many']", default_load: "Literal['raise', 'selectin', 'joined']", relationship: Any) -> None

index (property)#

index: int

SchemaMode#

Startup-selected mapping from logical schemas to PostgreSQL namespaces.

SchemaMode(kind: "Literal['single', 'isolated']", schema: str | None = None, central: str | None = None, isolation: "Literal['namespace', 'role']" = 'namespace') -> None

isolated (classmethod)#

isolated(*, central: str, isolation: "Literal['namespace', 'role']" = 'namespace') -> SchemaMode

single (classmethod)#

single(schema: str) -> SchemaMode

SchemaRef#

A logical or fixed schema reference compiled by a registry.

SchemaRef(kind: "Literal['fixed', 'central', 'tenant']", name: str | None = None) -> None

StorageSpec#

The fixed layout dimensions for one model.

StorageSpec(field_count: int, relation_count: int, basicsize: int) -> None

fingerprint_model#

fingerprint_model(schema: str, table: str, columns: tuple[ColumnSpec, ...], relationships: tuple[RelationshipSpec, ...], table_uniques: tuple[Any, ...] = (), table_indexes: tuple[Any, ...] = ()) -> bytes

A stable SHA-256 fingerprint of one compiled model.

The encoding is versioned and canonical: it never uses repr() or Python's randomized hash(), so a fingerprint is comparable across processes and runs. Table-level constraints append nothing when a model declares none, so a model without them fingerprints exactly as before.

fingerprint_registry#

fingerprint_registry(specs: tuple[ModelSpec, ...]) -> bytes

A stable deployment fingerprint over every model, in order.

wreath.orm.compiler#

Compile immutable query objects into parameterized PostgreSQL statements.

This runs once per query shape and the result is cached on the registry, so it is not on the per-row path; readable, auditable SQL generation is worth more here than speed. Values never reach the SQL text or the cache key.

Only semantics-preserving rewrites happen: duplicate projections collapse, primary keys are added because identity requires them, and duplicate load requests merge. PostgreSQL chooses scans and joins; Wreath chooses only the shape of the data and the number of round trips.

CompiledQuery#

CompiledQuery(sql: 'str', bind_values: 'tuple[Any, ...]', bind_oids: 'tuple[int, ...]', result_model: 'ModelSpec | None', selected_columns: 'tuple[ColumnSpec, ...]', load_plan: 'LoadPlan', shape_key: 'bytes', projected_columns: 'tuple[ColumnSpec, ...]' = ())

CompiledQuery(sql: str, bind_values: tuple[Any, ...], bind_oids: tuple[int, ...], result_model: ModelSpec | None, selected_columns: tuple[ColumnSpec, ...], load_plan: LoadPlan, shape_key: bytes, projected_columns: tuple[ColumnSpec, ...] = ()) -> None

JoinedStep#

A to-one relationship assembled from a LEFT JOIN in the same statement.

JoinedStep(relationship: RelationshipSpec, columns: tuple[ColumnSpec, ...], offset: int, alias: str, nested: tuple[JoinedStep, ...] = ()) -> None

LoadPlan#

LoadPlan(columns: 'tuple[ColumnSpec, ...]', joined: 'tuple[JoinedStep, ...]' = (), selectin: 'tuple[SelectinStep, ...]' = ())

LoadPlan(columns: tuple[ColumnSpec, ...], joined: tuple[JoinedStep, ...] = (), selectin: tuple[SelectinStep, ...] = ()) -> None

SelectinStep#

A relationship loaded by a second, batched statement.

SelectinStep(relationship: RelationshipSpec, nested: tuple[LoadOption, ...] = ()) -> None

SqlBuilder#

Accumulates SQL text and bind values for one statement.

Public because wreath._series renders its own statements against the same predicate machinery. Placeholder numbering is positional, so a caller must emit text and binds in the order they appear in the SQL.

SqlBuilder() -> None

bind#

bind(value: Any, oid: int) -> str

sql#

sql() -> str

text#

text(value: str) -> None

check_predicate_columns#

check_predicate_columns(model: type, node: Expression) -> None

Raise if a predicate names a column of some other model.

Select.where cannot check this, because a predicate is built before it meets a query. Nothing needs it per request either -- a handler's predicate is written next to the query it filters. A declared query is different: wreath.queries writes the predicate at class-definition time and runs it much later, so a mistyped model would surface as a broken statement on a request instead of at import. This is the walk that moves it back.

Relationship traversals are checked only at their first hop; the registry owns the rest of the path, and it is not available this early.

compile_count#

compile_count(registry: Any, select: Select) -> tuple[str, tuple[Any, ...], tuple[int, ...]]

Compile SELECT COUNT(*) for the rows select matches.

Reuses the filter-join and predicate machinery but drops projection, load joins, ordering, paging, and row locking -- none of them change how many parent rows match. Filter joins are always to-one (the compiler rejects to-many filter joins), so COUNT(*) over the joined-and-filtered rows equals the parent-row count; no subquery wrapper is needed.

Unlike compile_select this is not plan-cached: it is called at most once per page request, not per row, so rendering fresh -- which lets the bound values be captured directly, without the cached bind program -- is both simpler and cheap enough. Returns (sql, values, oids).

compile_declared_expression_values#

compile_declared_expression_values(expressions: Iterable[Expression], placeholder: type) -> Callable[[dict[str, Any]], tuple[Any, ...]]

Compile direct bind extraction for a fixed expression sequence.

Declared calculated views emit their predicates more than once inside one statement. They need the same parameter/literal program as a declared ORM query, without first manufacturing a Select merely to reach it.

compile_declared_values#

compile_declared_values(select: Select, placeholder: type) -> Callable[[dict[str, Any]], tuple[Any, ...]]

Compile direct parameter-to-wire extraction for a declared query.

Unlike the ordinary cached-plan binder, this program does not need a newly rebound Select: fixed literals come from the declaration and named placeholders read the call's value mapping directly. Parameter names are identifiers validated by Param; they enter generated source only through repr, while types and literal values stay in the closed namespace.

compile_rebind#

compile_rebind(node: Expression, placeholder: type, found: list[Any]) -> Callable[[Any], Any] | None

Compile a substitution program for one predicate holding placeholders.

A declared query (see wreath.queries) fixes its tree once and varies only the values in it, so the traversal that finds those positions can run at declaration time instead of per call. Returns a function that rebuilds the predicate from a mapping of parameter values, or None when the predicate holds no placeholders at all and can be reused as it stands. Every placeholder encountered is appended to found, in the order it binds, so the caller can name its parameters without a second walk.

This lives here rather than in wreath.queries because the shape of an expression tree is this module's knowledge: it has to stay in step with _append_bind_paths and _shape_expression, and the three are only reviewable together. A placeholder node's bind method owns what a parameter means; this function owns only where one may appear.

compile_select#

compile_select(registry: Any, select: Select) -> CompiledQuery

Compile select against registry, using its bounded plan cache.

conjoin#

conjoin(predicates: tuple[Predicate, ...]) -> Predicate

plan_filter_joins#

plan_filter_joins(registry: Any, spec: ModelSpec, expressions: Iterable[Expression], clauses: list[str]) -> dict[tuple[Any, ...], str]

Emit an INNER JOIN per relationship path expressions reach through.

These joins constrain rows and select nothing: filtering is not loading, so a filtered relation stays unloaded unless the query also .include()s it. INNER rather than LEFT because a parent with no matching child cannot satisfy a predicate on the child's column, and INNER lets PostgreSQL reorder the join.

Takes expressions rather than a Select because a calculated view groups by a column as well as filtering by one, and a GROUP BY on a related column needs the same join a predicate on it would get. Aliases are returned keyed by relationship path, which is what render_predicate reads.

qualified#

qualified(spec: ModelSpec) -> str

quote#

quote(identifier: str) -> str

Quote one registry-validated identifier.

Identifiers reach this function only after validate_identifier, so the embedded-quote escape is belt-and-braces rather than the actual defense.

render_predicate#

render_predicate(node: Expression, builder: SqlBuilder, alias: str, joins: dict[tuple[Any, ...], str], *, registry: Any = None) -> None

Render one predicate into builder, numbering placeholders positionally.

Args:

• node — The predicate tree to render.
• builder — Accumulates the SQL text and the bound values, in order.
• alias — The table alias the unqualified columns belong to.
• joins — Alias per relationship path, for columns reached through one.
• registry — Needed only to render an IN (SELECT ...) subquery, which has
• to resolve its own model to a table. Callers that never build one
• may leave it None; a subquery reached without it raises rather than
• rendering something half-qualified.

shape_of#

shape_of(registry: Any, select: Select) -> bytes

The keyed shape, walking the tree in Python for a node C cannot key.

orm_shape_configure hands the builder a fixed set of expression classes and it dispatches by exact type, so a node added since the extension was built raises ORMError("cannot key ...") rather than quietly producing a key that ignores it. That refusal is what makes the recovery safe: _shape_of_walk raises the identical error for anything genuinely unkeyable, so catching it either recovers a node the C does not know yet or re-raises exactly what the caller would have seen.

Every expression class defined by this release takes the native path. The recovery permits a Python expression class to precede an extension rebuild without letting a stale extension key two different queries identically. A version mismatch may therefore make compilation slower, never incorrect.

An ordering that is not a bare column walks before the call rather than through the recovery: the C keyer reads ordering.expression.column, which a distance node does not have, and that surfaces as an AttributeError -- not the ORMError this recovery is built on. Deciding it from a flag the Select already computed is one attribute read, and it keeps the contract above as narrow as it claims.

wreath.orm.introspection#

Startup schema validation.

Wreath never creates, alters, or drops database objects. It reads pg_catalog once at startup and reports, deterministically, where the database and the compiled models disagree.

information_schema is deliberately not used: it reports SQL-standard type names and loses the type OIDs and array element identity this comparison depends on.

Every catalog column is cast to a type the driver has a codec for. Wreath's PostgreSQL driver decodes a deliberately small set -- bool, the int and float widths, text/varchar, bytea, uuid, the date and timestamp types, numeric, json and jsonb -- and hands anything else back as the raw wire bytes. The catalog is made almost entirely of types outside that set: nspname, relname and attname are name, atttypid and typelem are oid, contype is "char", conkey is int2[] and indkey is int2vector. Read uncast, a column name arrives as b'id' and matches no declared column, so the validator reported every column of every table missing.

The casts are not cosmetic, and they are not only about arrays. The raw bytes also differ between the two result formats -- atttypid is b'20' on a cold operation (text) and b'\x00\x00\x00\x14' once the plan is cached (binary) -- so an uncast read gives a different wrong answer on the first call than on the second. Casting to text/bigint makes both formats decode to the same Python value, which is why this file does not need to care whether the statement is cold or warm.

ExtensionTypeResolution#

Where one extension type's OID came from, or that it has none.

ExtensionTypeResolution(type_name: str, extension: str, oid: int, schema: str, current_schema: str) -> None

installed (property)#

installed: bool

SchemaDiff#

SchemaDiff(issues: 'tuple[SchemaIssue, ...]')

SchemaDiff(issues: tuple[SchemaIssue, ...]) -> None

report#

report(limit: int = 50) -> str

SchemaIssue#

One disagreement, ordered so a diff is stable across runs.

SchemaIssue(schema: str, table: str, column: str, issue_code: str, detail: str) -> None

declared_extension_columns#

declared_extension_columns(registry: Any) -> tuple[tuple[Any, Any], ...]

Every (spec, column) in registry whose type comes from an extension.

declared_index_methods#

declared_index_methods(registry: Any) -> tuple[str, ...]

The access methods this registry declares that spell an operator class.

btree and GIN are excluded because their index descriptor carries no operator-class field at all, so nothing about them can disagree with the catalog over one. Sorted, so the probe below asks the same question in the same order on every run.

probe_default_opclasses#

async probe_default_opclasses(connection: Any, methods: tuple[str, ...]) -> dict[tuple[str, int], str]

Read {(access method, indexed type OID): default operator class}.

Split from resolve_default_opclasses for the same reason probe_extension_types is split from resolve_extension_types: the reading is useful without the binding, and a test that wants to know what this server's defaults are should not have to build a registry to find out.

An access method with no default operator class -- and a method this server has never heard of -- simply contributes no rows. There is nothing to raise about: it means every declared operator class on that method is explicit, which is the case the catalog already records verbatim.

probe_extension_types#

async probe_extension_types(connection: Any, wanted: dict[str, str]) -> tuple[ExtensionTypeResolution, ...]

Read one OID per {type_name: extension} entry, without deciding anything.

Split out from resolve_extension_types because wreath.doctor needs the same reading without the startup failure: a readiness report that raised would be a worse tool than the startup check it duplicates.

resolve_default_opclasses#

async resolve_default_opclasses(registry: Any, connection: Any | None = None) -> dict[tuple[str, int], str]

Give registry this database's default operator class per index method.

An index declared index_ops="vector_l2_ops" on ivfflat names pgvector's default operator class for that method, and PostgreSQL does not remember that it was named: pg_get_indexdef deparses it away, and wreath's catalog read deliberately records a default as the empty string so that an index declared without an operator class is not reported as drifted. Without this, the desired descriptor keeps saying vector_l2_ops and the catalog keeps saying nothing, so detect rediscovers drift on every run and generate emits a MANUAL forever for an index that is already exactly right.

The answer belongs to the database, so it is read from one, once per registry, and cached on it. A registry that declares no such index does no I/O at all. connection is for the migration entry points, which already hold one and are the only thing that builds a desired descriptor; passing none borrows from the registry's own pool instead.

resolve_extension_types#

async resolve_extension_types(registry: Any) -> tuple[ExtensionTypeResolution, ...]

Give this registry's extension-typed columns the OIDs this database uses.

Runs at startup, before anything binds a value. pgvector's vector is not a compile-time OID the way bigint is -- CREATE EXTENSION assigns it, and a different database assigns a different one -- so the OID is read here once and handed to the driver's codec table. Nothing resolves a type at request time; that is the whole point of doing it here.

A registry that declares no extension type does no I/O at all.

Raises:

• ExtensionNotInstalledError — a declared type has no OID in this database.
• The message names the extension, the schema wreath looked in, and
• the first column that wanted it, because the alternative is an
• unrecognised-OID error at the first query with none of that in it.

validate_registry#

async validate_registry(registry: Any) -> SchemaDiff

Compare a registry's models to the live catalog.

Runs on the registry's read pool when it has one, since validation is a read. Returns the diff; raises or warns according to validate_schema.

wreath.orm.validation#

One-pass validation from a request body straight into a model.

Without this there are two validation engines on the write path: request binding checks a JSON payload against a dataclass, and then every assignment to a model re-checks the same value through its column's PgType. The value is proven twice and copied twice.

Here a body is validated by the columns themselves, once, directly into the model's cells. The column type stays the single source of the type rules -- this does not add a second set -- but it runs exactly once per field, and the value that survives it is the value that gets bound to SQL.

The validator is generated per model, at route-compile time. A model's columns are fixed the moment the class exists, so the loop that walked them per request was re-deciding, on every field of every body, facts that were settled at import: what the field is called, whether it is nullable, whether it has a default. Generating the source unrolls all of that into straight-line code, and lets a cross-field rule read the values out of local variables instead of loading them back out of the cells they were just written to. See constraints.py for the same technique applied to a single column's checks.

The database still enforces its own structural constraints (types, NOT NULL, keys). That is deliberate: it is the backstop, not the validator. Application rules such as enums stay in Python, so the schema stays portable.

compile_model_validator#

compile_model_validator(model: type[Model]) -> Callable[[Any, tuple[Any, ...]], Any]

Build the validator for model; generated once, run per request.

wreath.orm.errors#

ORM error types.

Declaration mistakes surface as DeclarationError while a registry compiles, never at request time. Everything else is raised by a session.

DeclarationError#

An invalid model, column, or relationship declaration.

Always raised while app.orm() compiles a registry, so a running application never discovers a broken declaration mid-request.

DetachedInstanceError#

An operation required an object that a session still owns.

ExtensionNotInstalledError#

A column declares an extension type the database does not provide.

Raised at startup, while extension type OIDs are being resolved, so a Vector column on a database without CREATE EXTENSION vector fails where the message can name the extension and the schema -- not at the first query, where it would surface as an unrecognised OID.

ExtensionNotInstalledError(message: str, extension: str = '', schema: str = '') -> None

MappingError#

A result set does not match the model it was asked to hydrate.

MultipleResultsError#

fetch_one() matched more than one row.

NoResultError#

A required primary key or query matched no row.

ORMError#

Base class for every ORM error.

RegistryError#

Invalid registry configuration or lookup.

SchemaMismatchError#

The database schema disagrees with the compiled models.

SchemaMismatchError(message: str, diff: tuple[object, ...] = ()) -> None

SessionClosedError#

The session was used after it was closed.

SessionError#

Invalid session use.

UnloadedAttributeError#

A scalar column was read on an object that never loaded it.

Attribute access never issues SQL, so an unprojected column raises rather than silently returning None or fetching a row.

UnloadedRelationshipError#

A relationship was read without being loaded.

Load it explicitly with .include(...) on the query or await session.load(instance, Model.relationship).

StaleDataError#

A write matched no row, so the object it came from is out of date.

Raised when an UPDATE or DELETE built from a loaded object affects zero rows: another session deleted it, or changed the key it was found by. The driver may consider the statement successful, but the ORM requires exactly one matching row.

wreath.migrations#

Wreath-metal PostgreSQL migration configuration and bounded result views.

DeferredDeclarationError#

A deferred migration that could not be declared.

DowngradeHazard#

One ORM-mapped object a downgrade would strand or retype under the code.

DowngradeHazard(schema: str, table: str, name: str, kind: str, reason: str) -> None

explain#

explain() -> str

DowngradeWouldStrandCode#

A downgrade was refused because live ORM code still references what it removes.

DowngradeWouldStrandCode(schema: str, hazards: tuple[DowngradeHazard, ...]) -> None

DowngradeWouldStrandRecodedData#

A downgrade was refused because a re-encode has changed the values under it.

Reverse DDL can verify cleanly while leaving values encoded for the newer schema. This refusal requires the values to be converted back before the schema definition is reverted.

DowngradeWouldStrandRecodedData(schema: str, hazards: tuple[RecodedColumnHazard, ...]) -> None

MigrationBlockedByPass#

A migration was refused because a pass has not published for a column it narrows.

MigrationBlockedByPass(schema: str, hazards: tuple[PendingPassHazard, ...]) -> None

PendingPassHazard#

One column this artifact narrows while a pass is still converting it.

PendingPassHazard(schema: str, table: str, column: str, action: str, pass_name: str, tenant: str, phase: str, holes_open: int) -> None

explain#

explain() -> str

RecodedColumnHazard#

One column this downgrade touches whose values a re-encode has changed.

RecodedColumnHazard(schema: str, table: str, column: str, action: str, pass_name: str, tenant: str, phase: str, finished: bool, ledger_row_present: bool = True) -> None

explain#

explain() -> str

Recode#

Shape A: re-encode one column's values in place.

Args:

• column — the column being re-encoded.
• mapping — old value to new value. Finite and invertible, because that
• is exactly what lets the scan widen col == new into
• col IN (new, old) rather than merely permitting it. A conversion
• that cannot enumerate its pairs is not a Recode.
• chunk — rows per chunk, passed straight to the pass.
• pace — how much of the machine the walk may be.
• name — the pass's ledger identity; derived from the column by default.

The declaration lives beside the model rather than inside a migration file, because this repository generates migration artifacts from the catalog and has no authored-migration surface to hang a decorator on. See the guide.

Recode(column: Any, mapping: Any, chunk: int = 10000, pace: Any = None, name: str | None = None) -> None

build#

build() -> ChunkedPass

The walk that performs the conversion.

No gate: Shape A adds no column, so there is nothing for a later migration to narrow and no fact for it to wait on. The safety mechanism is scan, and it runs before the walk starts rather than after it finishes.

rewrites records the column association needed to refuse an unsafe downgrade. Unlike guards, it is never cleared: completing the pass does not restore the previous encoding.

converts (property)#

converts: str

schema.table.column for the column being converted.

pass_name (property)#

pass_name: str

scan#

scan(**populations: Any) -> ScanReport

Every read of this column, classified. See ._migrations.scan.

Retype#

Shape B: change one column's type by draining it into a new one.

Args:

• column — the column being drained -- **the one a later migration will
• narrow**, which is what the published fact names.
• into — the new column's name, added by this migration's own DDL as
• nullable. Nullable-then-not is a change the ORM already models.
• using — the SQL expression producing the new value.
• chunk — rows per chunk, passed straight to the pass.
• pace — how much of the machine the walk may be.
• name — the pass's ledger identity; derived from the column by default.

The gate is not optional here and it is the half X1 cannot ship without. Verification is NOT VALID then VALIDATE CONSTRAINT -- the constraint the swap migration is about to add -- so the thing proven and the thing later enforced are the same predicate, and a walk that was subtly wrong cannot verify its own bug.

Retype(column: Any, into: str, using: Any, chunk: int = 10000, pace: Any = None, name: str | None = None) -> None

build#

build() -> ChunkedPass

The walk, its gate, and the fact the gate publishes.

Ceiling.at_launch() and scope="pass" are not a choice: a gate with scope="unit" cannot publish a fact, and a whole-pass gate is refused on a recurring pass, so a pass that guards a column is necessarily a bounded one that finishes.

constraint_name (property)#

constraint_name: str

converts (property)#

converts: str

pass_name (property)#

pass_name: str

publishes (property)#

publishes: str

The fact a later migration waits on before dropping the old column.

scan#

scan(**populations: Any) -> ScanReport

Shape B has no re-encode window, so there is nothing to scan.

Both columns are separately typed and correct throughout; an unconverted row reads NULL, which the ORM already models and the gate proves gone. Returning an explicitly empty report rather than raising keeps one calling shape for both declarations.

ScanReport#

Everything the scan looked at, and what it concluded.

examined distinguishes a clean result from a scan that inspected no declarations. scanned_nothing makes that absence of evidence explicit.

ScanReport(column: str, examined: int, hazards: tuple[TransitionalHazard, ...] = (), rewrites: tuple[TransitionalHazard, ...] = (), scanned_nothing: bool = False, shape: str = 'recode') -> None

blocking (property)#

blocking: tuple[TransitionalHazard, ...]

explain#

explain() -> str

waived (property)#

waived: tuple[TransitionalHazard, ...]

TransitionalHazard#

One read that cannot be proven safe for the conversion window.

TransitionalHazard(site: str, column: str, operation: str, verdict: str, detail: str, rewrite: str | None = None, waiver: str | None = None) -> None

explain#

explain() -> str

waived (property)#

waived: bool

TransitionalContractUnproven#

A deferred migration whose reads could not be proven safe for the window.

The mirror of MigrationBlockedByPass, one step earlier: that one refuses a migration for narrowing a column a pass has not finished, this one refuses to start the pass while a read of that column would silently mean something else while it runs.

Refusal is the default and the asymmetry is the argument -- a false refusal costs an argument with the tool, a false permission costs a data incident that surfaces a week later.

TransitionalContractUnproven(report: Any) -> None

FleetResolution#

FleetResolution(current: 'int', apply: 'int', verify: 'int', ambiguous: 'int', blocked: 'int')

FleetResolution(current: int, apply: int, verify: int, ambiguous: int, blocked: int) -> None

total (property)#

total: int

MigrationConfig#

MigrationConfig(database: 'str', policy: 'ResolutionPolicy', catalog_chunk_size: 'int' = 256, concurrency: 'int' = 8, max_failures: 'int' = 100)

MigrationConfig(database: str, policy: ResolutionPolicy, catalog_chunk_size: int = 256, concurrency: int = 8, max_failures: int = 100) -> None

MigrationBaseline#

A reviewed zero-operation root adopting one existing live schema.

artifact is a normal checksummed WMA1 artifact, but its source and target fingerprints are identical and all three operation tapes are empty. Applying it therefore records history and executes no DDL. descriptor is the bounded WMD1 catalog inventory used for human review.

MigrationBaseline(artifact: NativeMigrationArtifact, fingerprint: bytes, descriptor: bytes, object_count: int) -> None

MigrationDetection#

Bounded result of comparing ORM intent with one live schema.

MigrationDetection(desired_fingerprint: bytes, actual_fingerprint: bytes, diff: NativeMigrationDiff) -> None

current (property)#

current: bool

MigrationGeneration#

Review metadata retained by the native named operation planner.

MigrationGeneration(desired_fingerprint: bytes, actual_fingerprint: bytes, diff: NativeMigrationDiff, plan: NativeMigrationPlan, sql: NativeMigrationSql) -> None

MigrationApplyResult#

Verified result of one locked, transactional single-schema application.

MigrationApplyResult(migration_id: bytes, checksum: bytes, source_fingerprint: bytes, target_fingerprint: bytes, destructive_approved: bool) -> None

MigrationRevertResult#

Verified result of one locked, transactional single-schema downgrade.

MigrationRevertResult(migration_id: bytes, checksum: bytes, source_fingerprint: bytes, target_fingerprint: bytes, destructive_approved: bool, forced: bool) -> None

NativeCatalogSnapshot#

Canonical image plus its bounded native names/signatures descriptor.

NativeCatalogSnapshot(image: bytes, descriptor: bytes) -> None

NativeMigrationArtifact#

Verified immutable artifact metadata with its packed operation tape.

NativeMigrationArtifact(data: bytes, checksum: bytes, migration_id: bytes, parent_checksum: bytes, source_fingerprint: bytes, target_fingerprint: bytes, operation_tape: bytes, named_plan: bytes, sql_tape: bytes) -> None

NativeMigrationChain#

Metal-verified artifact-chain tip and migration count.

NativeMigrationChain(checksum: bytes, target_fingerprint: bytes, migration_count: int) -> None

NativeMigrationDiff#

A deterministic packed operation tape produced entirely by Wreath-metal.

NativeMigrationDiff(operation_count: int, tape: bytes) -> None

NativeMigrationPlan#

Deterministic named WMP1 operation plan produced by Wreath-metal.

NativeMigrationPlan(operation_count: int, tape: bytes) -> None

NativeMigrationSql#

Deterministic WMS1 statement tape derived from a named native plan.

NativeMigrationSql(operation_count: int, manual_count: int, destructive_count: int, tape: bytes) -> None

ResolutionPolicy#

Whether fleet readiness trusts verified history or audits every catalog.

ResolutionPolicy(kind: "Literal['managed', 'strict']", sample_size: int = 0) -> None

managed (classmethod)#

managed(*, sample_size: int = 0) -> ResolutionPolicy

strict (classmethod)#

strict() -> ResolutionPolicy

TenantState#

One tenant's recorded migration state, as the directory knows it.

status is one of the HISTORY_* codes. VERIFIED means the checksummed history was proven and can be trusted as the fast readiness authority; UNKNOWN forces catalog verification; AMBIGUOUS and BLOCKED are terminal operational states the resolver never treats as current. Applications build these from their own tenant directory; the runner never invents tenant identity.

TenantState(tenant_id: int, migration: int, checksum: int, generation: int, status: int = 0) -> None

apply_single_artifact#

async apply_single_artifact(registry: Any, connection: Any, artifact_data: bytes, *, allow_destructive: bool = False) -> MigrationApplyResult

Apply one authoritative artifact under a transaction-scoped native plan.

adopt_single_baseline#

async adopt_single_baseline(registry: Any, connection: Any, artifact_data: bytes) -> MigrationApplyResult

Record one zero-operation root after re-verifying code and catalog.

The only DDL this may issue is creation of Wreath's own migration-history schema/table. It never builds or executes an application DDL block. A prior history row, non-root artifact, operation, or fingerprint mismatch is refused inside the same advisory-locked transaction that records adoption.

connect_migration#

async connect_migration(dsn: str) -> Any

Open a dedicated Wreath-metal migration connection.

detect_single#

async detect_single(registry: Any, connection: Any) -> MigrationDetection

Compare one compiled single-schema registry with live PostgreSQL.

generate_single_plan#

async generate_single_plan(registry: Any, connection: Any, *, fleet: bool = False) -> MigrationGeneration

Build a deterministic named plan for one live physical schema.

fleet=True fingerprints the catalog with every tenant-local schema name blanked, so the artifact verifies against any structurally identical tenant rather than only the one it was generated from. That is the whole reason one artifact can cross a fleet: nspname is the first column of every catalog branch, so two byte-identical tenants otherwise fingerprint differently and an artifact could never match the second one.

A fleet artifact and a single-schema artifact are therefore not interchangeable, and cannot be confused for one another: their fingerprints differ, so applying the wrong kind fails the source-fingerprint refusal rather than migrating anything.

generate_single_baseline#

async generate_single_baseline(registry: Any, connection: Any, *, migration_id: bytes) -> MigrationBaseline

Adopt current PostgreSQL state as a root without replaying its history.

The live catalog and declared ORM image must match exactly. Drift is never folded into a baseline: doing that would bless a model that does not describe the database and make the first later migration plan relative to the wrong state. The returned artifact contains no operation and therefore cannot execute DDL when applied.

pack_tenant_directory#

pack_tenant_directory(tenants: Any) -> bytes

Pack tenant-directory rows into the resolver's contiguous snapshot.

The layout is the one the native metal resolver consumes; keeping the pack here (rather than in each caller) means the wire format has exactly one author.

resolve_fleet#

resolve_fleet(tenants: Any, *, target_migration: int, target_checksum: int, directory_generation: int) -> FleetResolution

Classify a whole tenant fleet's readiness in one metal invocation.

This is the managed fleet runner: it packs the caller's tenant directory into the native snapshot and resolves it with a single Wreath-metal call, returning bounded per-bucket counts rather than one Python object per tenant. A tenant is current only when its trusted, verified history is at the target migration and checksum for the current directory generation; unknown, stale, or wrong-generation tenants fall to verify (catalog audit), and ambiguous/blocked stay terminal.

The runner reads no live database and acquires no DDL authority. Turning a verify count into per-tenant catalog audits is a separate, explicitly privileged step; classification is deliberately side-effect free.

revert_single_artifact#

async revert_single_artifact(registry: Any, connection: Any, artifact_data: bytes, *, allow_destructive: bool = False, force: bool = False) -> MigrationRevertResult

Undo one authoritative artifact: the exact inverse of apply_single_artifact.

The named plan is inverted in metal (every add becomes a drop, every signature swaps), so the downgrade tape is derived from the same authority the upgrade was — never guessed. The artifact must be the current history tip; the transaction verifies the live catalog matches the artifact target, runs the reverse DDL block, requires the catalog to return to the artifact source, deletes the tip history row, and commits — or rolls everything back.

Unless force is set, the downgrade is refused when the running ORM still maps a column or table the reverse would drop (or a type it would change): downgrading production under code that still references those objects strands the deployed code. force exists for the legitimate case of rewinding a local stack to re-migrate.

scan_transitional_reads#

scan_transitional_reads(declaration: Any, *, registry: Any = None, queries: Any = (), views: Any = (), strict: bool = True) -> Any

Classify every read of the column declaration converts.

Returns the report. With strict, raises TransitionalContractUnproven when anything is unproven -- including when nothing was scanned at all. An empty scan is absence of evidence, not proof that reads remain valid during the transition.

transitional_read#

transitional_read(column: Any, *, reason: str) -> Any

Waive the transitional check for one read, with a written reason.

The reason is not decoration. It appears in wreath migrations check output as a count and in --json in full, so "we waived everything" is visible in review rather than discovered afterwards.

This form decorates a function. A declared query is a slotted value with nowhere to hang an attribute, so it takes waive_transitional instead; the error below says so rather than failing obscurely.

unpack_catalog_descriptor#

unpack_catalog_descriptor(descriptor: bytes) -> list[dict[str, Any]]

Decode a bounded WMD1 catalog inventory into reviewable values.

This is an operator view, not migration execution input. Execution always consumes the native image and verified artifact; editing this list changes nothing about what can be adopted.

waive_transitional#

waive_transitional(column: Any, *, site: str, reason: str) -> None

Waive one read named by its declared site, for reads that are values.

A QueryDeclaration is immutable and slotted, so there is nothing to decorate. Keying on the declared name rather than on object identity is deliberate: the name is the thing the declaration sells, it is stable across a reload, and id() is not an identity -- a lesson this codebase has paid for twice.

wreath.schema#

The wreath schema: the tables wreath owns, and how they get there.

Wreath's subsystems -- the job queue, the message bus, sessions, rate limits, idempotency, the webhook inbox and outbox, the pass ledger, series buckets -- need tables of their own. Those tables are not the application's data model and they do not belong in the application's migration artifact: the artifact describes what the author declared, and nobody declared a job queue.

So they live in a schema of their own, named wreath by default, and this module puts them there. A framework user who does nothing gets a working database. A framework user whose role cannot CREATE SCHEMA is told exactly what to hand a DBA, and is refused at startup rather than at the first enqueue.

Three properties hold this together, and each is load-bearing:

• The user's migration diff cannot see these tables. The catalog read is scoped to one named schema (migrations.py), so a diff of the application's schema never observes wreath.* and never proposes dropping it. Separation is by construction, not by a filter someone has to maintain.
• Every reference is fully schema-qualified. Nothing here resolves through search_path, which is what lets it compose with an isolated tenant session -- that binds search_path to the tenant's schema alone.
• An upgrade step must be safe for the previous version to keep running against. A fleet mid-rollout has two wreath versions live on one database. Steps are therefore additive: add a table, add a nullable or defaulted column, add an index. A column stops being read in one release and is dropped two releases later, once no supported version reads it. That gap is what makes a rollback survivable.

Read ~/code-maps/designs/35-wreath-schema.md for the reasoning, the rejected alternatives, and the staging.

Component#

A subsystem's claim on the wreath schema.

Args:

• name — Stable identifier, used as the version-marker key and the advisory
• lock key. Renaming one strands its recorded version, so treat it as
• part of the on-disk format.
• steps — Ordered steps. Step 1 creates the component's tables.
• relations — The relations that must exist for the component to work,
• unqualified. Used by verify to say precisely what is missing when
• wreath is not managing the schema.
• schema — The schema the component's tables live in.

Component(name: str, steps: tuple[Step, ...], relations: tuple[str, ...] = (), schema: str = 'wreath') -> None

qualified (property)#

qualified: bool

Whether this component's tables live in schema.

Not every one does, and the difference is history rather than design. Six components name their tables "wreath"."jobs"; five predate that and use an unqualified wreath_-prefixed name, which PostgreSQL resolves through search_path -- normally public. Moving them is a non-additive change: a worker on the previous version looks for wreath_session and would not find it, which is precisely what the additive rule exists to prevent. So they are registered where their rows actually are, and a move is a later, staged concern.

sql#

sql(*, from_version: int = 0, create_schema: bool = True) -> str

The component's DDL as one semicolon-joined script.

This is a derivation, not a source. Step.statements holds the DDL; this joins it. Every schema_sql() in wreath returns this, so the tuple and the script cannot disagree -- which they did while jobs carried both, spelled out twice.

Args:

• from_version — Emit only steps above this version.
• create_schema — Prepend CREATE SCHEMA IF NOT EXISTS. A component
• whose tables are unqualified has no schema to create, and
• emitting one would create a schema nothing then uses.

Returns:

• SQL, one statement per ;\n, with a trailing ;\n. Callers that
• execute it must split on ;\n -- or better, use statements(),
• which is why the tuple exists.

statements#

statements(*, from_version: int = 0) -> tuple[str, ...]

Every statement of every step above from_version, in order.

target_version (property)#

target_version: int

The version a database is brought to once every step has applied.

SchemaContributor#

A registered subsystem that declares one wreath-owned schema component.

SchemaContributor(*args, **kwargs)

component#

component() -> Component

SchemaError#

A wreath-owned schema is missing, unmanageable, or inconsistent.

SchemaNotManaged#

A required relation is absent and wreath was told not to create it.

Raised at startup, never at first use. A subsystem that registers is a subsystem that will be used, so deferring the check to the first enqueue reproduces exactly the failure this module exists to remove: a missing table surfacing as a runtime error, far from the configuration that caused it.

Step#

One ordered, additive change to a component's tables.

statements is a tuple rather than a semicolon-joined string on purpose. The driver speaks the extended query protocol exclusively, so it prepares every statement and PostgreSQL refuses cannot insert multiple commands into a prepared statement. Four call sites in this repository had each grown the same sql.split(";\n") workaround to get around a schema_sql() that returned a blob; a tuple of statements removes the workaround rather than repeating it a fifth time.

Args:

• version — The version this step brings the component to. Steps apply in
• ascending order and a component's first step is version 1.
• statements — The statements to execute, in order, each on its own.

Step(version: int, statements: tuple[str, ...]) -> None

bootstrap#

async bootstrap(database: Any, components: Sequence[Component], *, schema: str = 'wreath', workload: str = 'write', manage: bool = True) -> dict[str, int]

Bring the wreath schema up to date, or refuse and say what is missing.

Idempotent and safe to run concurrently: the work for each component happens under an advisory lock keyed on that component, so a fleet of workers starting together serialises rather than racing. Locking per component rather than globally means an unrelated subsystem's upgrade does not block a worker that does not use it.

Each step applies in its own transaction together with the version row that records it, so a crash mid-bootstrap leaves a database that describes itself accurately rather than one whose marker overstates what landed.

Args:

• database — A started wreath.postgres.Database.
• components — The components to bring up to date.
• schema — The schema to manage.
• workload — The pool to use. Must be a primary; DDL cannot run on a replica.
• manage — When False, create nothing and verify instead.

Returns:

• The version of each component after the run.

Raises:

• SchemaNotManaged — manage is False and a required relation is absent.

emit_sql#

emit_sql(components: Sequence[Component], *, schema: str = 'wreath', from_version: int = 0) -> str

The full ordered DDL, for a DBA to apply by hand.

This is the supported spelling for a deployment whose application role cannot CREATE SCHEMA. It is the same statements bootstrap would run, in the same order, so what a DBA applies and what wreath would have done cannot drift.

Args:

• components — The components to emit, in registration order.
• schema — The schema to emit for.
• from_version — Emit only steps above this version, for a database that is
• already partly upgraded.

Returns:

• Executable SQL, one statement per ; on its own line.

marker_statements#

marker_statements(schema: str = 'wreath') -> tuple[str, ...]

DDL for the schema and its version marker. Idempotent, applied first.

Per-component versions, not one global counter: a global number forces unrelated subsystems into a single order, so a jobs change and a webhooks change collide for no reason and a deployment using neither still has to move it.

verify#

async verify(database: Any, components: Sequence[Component], *, schema: str = 'wreath', workload: str = 'write') -> None

Refuse if any component's relations are absent. Never creates anything.

Raises:

• SchemaNotManaged — A required relation is missing. The message names the
• component, the relation, and the command that emits the DDL.

wreath.sql#

SQL that cannot carry an injection, because the values never enter the text.

Every SQL injection has the same shape: a value the caller supplied was turned into syntax instead of being sent as data. The defence has been known for decades -- bind the value, do not format it -- and applications keep getting it wrong anyway, because the two spellings look almost identical:

sql = f"SELECT id FROM shipments WHERE reference ILIKE '%{needle}%'"   # a hole
sql = "SELECT id FROM shipments WHERE reference ILIKE $1"              # not

The first one is shorter, it reads better, it is what an f-string is for, and nothing about it is flagged by the language, the type checker, or the driver. It is a defect that has to be caught by a person noticing, every time, forever.

Python 3.14 finally makes the difference visible to the machine. A t-string (PEP 750) is not a string: it evaluates to a string.templatelib.Template that keeps the literal parts and the interpolated values apart, exactly as the compiler saw them. So a template can be turned into $1-style SQL mechanically, and the interpolated values can only ever leave as parameters:

from wreath.sql import Identifier, Statement

rows = await db.raw(
    t"SELECT id, reference FROM {Identifier(schema, 'shipments')} "
    t"WHERE org_id = {org} AND reference ILIKE {pattern}"
).fetch()

org and pattern become $1 and $2. There is no argument the caller can pass -- no quote, no comment marker, no UNION -- that changes the statement, because their text is never concatenated into it. Changing that t back to an f does not quietly reintroduce the hole either: the result is a str, and Statement refuses a str with a message naming the fix.

Three interpolated types are spliced into the text rather than bound, because PostgreSQL resolves them at parse time and a parameter cannot stand in:

• Identifier -- a schema, table, or column name, double-quoted and escaped.
• Fragment -- literal SQL, the audited escape hatch (see its docstring).
• another Template or Statement -- a nested clause, whose own parameters are renumbered into the outer statement, so clauses compose.

Related: wreath.hardening is the rule set that finds the f-string spelling in an application's own source, at boot and in CI, and wreath.orm.Session.raw is the usual place a Statement is handed to the database.

Fragment#

Literal SQL spliced into a statement without quoting or binding.

This is the escape hatch, and it is deliberately the only one. Some SQL is genuinely syntax rather than data -- ASC, NULLS LAST, a whole ORDER BY clause assembled from an allow-list -- and no amount of binding can express it.

A Fragment is not a defence. Its whole purpose is to put caller-influenced text into the statement, so the value must come from the application, not from the request: look the caller's choice up in a mapping the application owns and pass the mapped value.

DIRECTIONS = {"asc": "ASC", "desc": "DESC"}
order = Fragment(DIRECTIONS[requested])       # KeyError, not an injection

Nothing about a Fragment is checked, and that is why it is spelled as a type rather than allowed implicitly: it is one greppable word that says "this text is SQL on purpose", so a review can find every one of them in a tree and read the mapping behind it.

Fragment(text: str) -> None

Identifier#

A schema, table, or column name, quoted for interpolation into SQL.

A parameter cannot name a relation: PostgreSQL resolves identifiers while it parses, long before a bind value exists. So an application that computes a table or schema name has to put it in the text, and this is the safe way to do it -- the name is double-quoted, any embedded quote is doubled, and a NUL is refused outright.

Several parts qualify a name: Identifier("northwind", "shipments") renders "northwind"."shipments".

Quoting makes a hostile name harmless rather than rejected; it becomes a relation that does not exist, and the statement fails with an ordinary "relation does not exist" instead of running something else. Deciding which names a caller may reach is a separate question, and one this class deliberately does not answer -- look the name up in a directory the application owns, as wreath.orm.TenantContext does.

Identifier(*parts: str) -> None

text (property)#

text: str

The quoted SQL for this name.

Statement#

SQL text with $1-style placeholders, compiled from a t-string.

text is what goes to the server and args is what is bound to it. The split is decided by the compiler, from the source, so no runtime value can move itself from one side to the other.

Statement(template: Template) -> None

wreath.store#

One keyed store, declared once, backed by memory or by PostgreSQL.

A rate limiter, an idempotency ledger, and a session table are three different features that need the same small thing: values under a key, aged by a timestamp, in one worker's memory or in a table every worker shares. Written three times, the PostgreSQL half is re-derived three times -- and the six re-derivations are exactly the parts that are easy to get subtly wrong:

• the table name reaches SQL by interpolation, so it must be a plain identifier;
• the schema is offered (Keyed.schema_sql) and never applied, because schema changes belong in the migration history with the rest of the schema;
• statements are prepared lazily, because a store is built while the application is being described and the database is not up yet;
• a claim is one INSERT ... ON CONFLICT ... RETURNING, never a read followed by a write;
• the clock is clock_timestamp() and never now(), which is frozen at transaction start;
• expired rows are dropped by an explicit PostgresStore.purge, never by a background thread.

Declare the shape once and pick a backend:

declaration = Keyed(
    table="wreath_session",
    columns=(Column("data", "jsonb", null=False),),
    key="sid",
    prefix="wreath_session",
)
store = PostgresStore(database, declaration)

What is not shared is the payload and the semantics: what the columns mean, whether the deadline is fixed per store or supplied per write, and whether the rows are claimed at all. Each caller keeps those.

CLAIMED (value)#

CLAIMED: _Claimed = CLAIMED

Sql#

A fragment the caller is asserting is safe to interpolate.

Column names are checked as identifiers; the expressions beside them cannot be, because an expression is arbitrary SQL by definition. The type is the check: a plain str in an expression position is refused, so text that arrived from a request cannot reach the statement without somebody having written Sql(...) around it and meant it.

A bind placeholder ($1) needs no marker -- it is the safe form, and requiring ceremony for it would push callers toward the unsafe one.

Column#

One payload column: its name, its SQL type, and whether it may be NULL.

Keyed#

The declaration of a keyed store: one key, some payload, one timestamp.

A last-touched store (deadline=False) expires nothing: the stamp is arithmetic for whoever reads it, and only an idle purge ages a row out.

claim=True requires a deadline -- there is nothing to reclaim without one -- and every payload column nullable, because a claim resets the payload.

An index on the stamp is what keeps purging a large table cheap, and the prefix is what keeps two stores over different tables from colliding on one prepared-statement name.

Args:

• table — the backing table. Interpolated, so it must be a plain identifier.
• columns — the payload. Their meaning belongs to the caller.
• key — the primary-key column.
• stamp — the timestamptz column rows are aged by.
• deadline — whether stamp is a deadline or a last-touched mark.
• ttl — seconds a row lives. None means the caller supplies it per write.
• index_stamp — also declare an index on stamp.
• claim — build the atomic claim statement.
• prefix — prepended to prepared-statement names.

Keyed(table: str, columns: tuple[Column, ...] = (), key: str = 'key', stamp: str = 'expires', deadline: bool = True, ttl: float | None = None, index_stamp: bool = False, claim: bool = False, prefix: str = 'wreath_store') -> None

schema_claim#

schema_claim(name: str) -> Any

This declaration's claim on the wreath schema, under name.

name is the caller's because one Keyed shape backs three different subsystems -- sessions, rate limits, idempotency -- and each needs its own version marker and its own advisory lock. Sharing one would make an upgrade to sessions block a worker that only uses rate limits.

Not component(). That name belongs to the zero-argument protocol Wreath.schema_components calls, and this needs an argument, so an object that reached the walk holding this method would have raised TypeError rather than contributing a claim. Two layers, two names.

schema_sql#

schema_sql() -> str

DDL for the backing table, semicolon-joined.

A derivation of statements(). Wreath applies these itself during lifespan now; this is retained for a caller applying the DDL by hand, and wreath schema sql is the supported spelling.

statements#

statements() -> tuple[str, ...]

DDL for the backing table, one statement per element.

The table name is unqualified, so it lands in whatever search_path resolves to. That is where every deployment's rows already are, and moving them into the wreath schema would not be additive -- a worker on the previous version would look for the old name. See wreath.schema.Component.qualified.

MemoryStore#

The same keyed store in one worker's memory: bounded, TTL'd, synchronous.

Enough on its own for a single-worker deployment or a sticky load balancer. Behind anything else it is a fast path in front of a shared store rather than a substitute for one, because a second worker's memory knows none of this.

Being synchronous is the feature, not an implementation detail: it is what makes claim atomic. There is no await between the read and the write, so no other task on this loop can interleave -- the in-process counterpart of the single statement PostgresStore.claim uses.

Nothing needs purging: entries expire lazily when read and max_entries bounds whatever is never read again.

The window opens when a key is first written, and writing again does not move it. A store that restarted the clock on every write would let a slow holder extend its own key indefinitely -- an idempotency key would outlive its TTL for as long as the handler kept touching it.

Over PostgreSQL the same rule is enforced a statement at a time rather than by this class: a caller's own upsert omits the stamp from its DO UPDATE, the way middleware.idempotency.PostgresIdempotencyStore's store statement does. The generated PostgresStore.claim is the deliberate exception -- its DO UPDATE does reset the stamp, because it only fires on an already-expired row and a reclaimed row must look exactly like a fresh one. Matching here is what lets one caller swap the backends and have a key honoured for the same length of time.

MemoryStore(*, ttl: float | None = None, max_entries: int = 4096, clock: Callable[[], float] = <built-in function monotonic>) -> None

claim#

claim(key: str) -> bool

Take ownership of key, or report that it is already held.

read#

read(key: str) -> Any

The stored value, CLAIMED when claimed but unwritten, or None.

set#

set(key: str, value: Any) -> None

Store value under key, keeping the deadline the key already has.

A key that is absent or already past its deadline starts a fresh window; one that is live keeps the deadline it was claimed or first written with.

PostgresStore#

A keyed store in a table every worker shares.

Holds the six disciplines named in this module's docstring, and generates the statements a keyed store always wants -- claim, read, delete, purge -- from the declaration. Anything shaped by the caller's payload is written by the caller and registered with define, which gets it the same lazy preparation and the same statement naming.

PostgreSQL owns the clock. Every generated statement reads clock_timestamp(), so workers on disagreeing wall clocks cannot disagree about when a row expires, and clock_timestamp rather than now() because now() is fixed at transaction start -- inside a transaction it would freeze time for as long as the transaction runs.

Generated reads go to the write pool by default, because a store that claims must read back from the primary that accepted its claim. A store that only reads may pass read_workload="read".

Args:

• database — a wreath.postgres.Database.
• declaration — what the store holds.
• read_workload — the pool generated reads go to. "write" by default.

PostgresStore(database: Any, declaration: Keyed, *, read_workload: str = 'write') -> None

claim#

async claim(key: str) -> bool

Take ownership of key, or report that someone else has it.

One INSERT ... ON CONFLICT (key) DO UPDATE ... WHERE expired RETURNING. A row comes back only when the insert succeeded or an expired row was reclaimed, so "a row came back" is the claim -- no owner column, no second round trip, and no window in which two workers both proceed. A read followed by a write would let both of them conclude they were first, which is the one thing a claim exists to prevent.

declaration (property)#

declaration: Keyed

The Keyed this store was built from. Frozen, so it is safe to share.

What a purge pass is handed: everything needed to walk the table -- its name, its key, its stamp -- with no reference back to this store or its connection pool.

delete#

async delete(key: str) -> str

Drop key. Not an error when it is already gone.

expired (property)#

expired: Sql

The predicate for a row the store no longer honours.

live (property)#

live: Sql

The predicate for a row that is still good.

The exact complement of expired, so a purge can never drop a row a read would still have honoured.

purge#

async purge(idle_seconds: float | None = None) -> str

Drop rows the store no longer honours.

Nothing calls this for you: a background thread would duplicate across workers and swallow its own failures. Run it from a durable job.

With idle_seconds, drops rows untouched for that long instead -- which is the only way to age out a last-touched store, whose stamp is never a deadline.

purge_count#

async purge_count(idle_seconds: float | None = None) -> int | None

purge, reporting how many rows went.

None when the driver's status cannot be read -- a test double, a backend that reports nothing. Splitting this from purge keeps the status string available to callers that want it while giving the scheduled job that runs this something it can actually record: "the purge ran" and "the purge removed 40 000 rows" are different facts.

read#

async read(key: str, *, live: bool = False) -> Any

The payload columns for key, or None.

live=True refuses a row whose deadline has passed, for a store that purges lazily -- which is every store, since nothing purges in the background.

schema_claim#

schema_claim(name: str) -> Any

This store's claim on the wreath schema, under name.

schema_sql#

schema_sql() -> str

DDL for the backing table, semicolon-joined.

table (property)#

table: str

The backing table's name, already checked as a plain identifier.

upsert#

upsert(*, values: Mapping[str, str], update: Mapping[str, str], where: str | None = None, returning: str | None = None) -> str

One INSERT ... ON CONFLICT DO UPDATE over this store's key.

Both mappings are column -> expression. The columns are interpolated, so each is checked as an identifier; the expressions are checked as Sql -- a bind placeholder, or a fragment the caller marked deliberately. The result is a single statement by construction -- two statements would be two chances for a concurrent worker to interleave.

window#

window(seconds: float | str | datetime.timedelta | None = None) -> Sql

A deadline seconds from the database's clock, not the caller's.

A string lifetime must be a bind placeholder; see Sql.

define (inherited from _Statements)#

define(name: str, sql: str, *, workload: str = 'write') -> None

Register sql under name for lazy preparation.

sql (inherited from _Statements)#

sql(name: str) -> str

The text registered under name. Useful when explaining a plan.

statement (inherited from _Statements)#

statement(name: str) -> Any

The prepared statement for name, preparing it on first use.

workload (inherited from _Statements)#

workload(name: str) -> str

The pool the statement registered under name runs against.

rows_affected#

rows_affected(status: Any) -> int | None

The row count in a PostgreSQL command tag, or None when unreadable.

A tag is DELETE 5, UPDATE 3, or -- the one that catches people -- INSERT 0 5, where the first number is a legacy OID and the second is the count. So the count is the last field, never the second, and four separate readers in this tree disagreed about that: two took the second field and returned None for every INSERT, which is indistinguishable from "the driver reported nothing" at the call site.

None rather than 0 when it cannot be read, because "no rows matched" and "this backend does not say" are different facts and a caller recording the first when it meant the second is reporting a clean sweep that never happened. A test double or a backend that reports nothing is a supported caller, not a broken one.

sql_identifier#

sql_identifier(value: str, *, what: str = 'table') -> str

Return value if it is a bare SQL identifier, else raise.

Table and column names are interpolated into statement text (they cannot be bound as parameters), so the only safe policy is to refuse anything that is not a plain identifier rather than to quote it.

wreath.queries#

Named, compiled reads over one model.

Every application grows a hand-written data-access layer: a module of functions that build the same five queries, with the filtering inlined at each call site and no shared name for the query that fetches a paddock's llamas. Wreath already owns the hard half of doing better -- a plan cache keyed by query shape -- and until now used it only internally.

A Queries class gives those reads names:

class Llamas(Queries[Llama]):
    by_paddock = query(Llama.paddock_id == Param("paddock"))
    overdue = query(Llama.checked_at < Param("before")).order_by(Llama.checked_at)

herd = await Llamas(session).by_paddock(paddock=7)

Two properties follow from declaring rather than building:

• The shape is fixed at class-definition time and only values vary, so each declaration compiles exactly once through the registry's existing plan cache, however many times it runs and whatever it is called with. There is no second cache here; a bound declaration is an ordinary Select whose plan-cache key is byte-identical to the hand-written query's.
• Mistakes move to import time. A column that belongs to another model, a parameter written somewhere a value cannot be bound -- those fail when the class is defined, not on the request that first runs them.

Declared searches also fuse. A vector distance and a relevance score are numbers on different scales, and no weighting between them survives a change of embedding model — but a rank is a rank, so fuse merges two named searches by where each row placed rather than by what either scored it:

class Documents(Queries[Document]):
    nearest = query().order_by(D.embedding.cosine_distance(Param("q"))).limit(50)
    matching = query(D.search.matches(Param("terms"))).order_by(...).limit(50)
    hybrid = fuse(nearest, matching).limit(10)

Reads only, deliberately. Writes already have an owner: the Session, whose one-way direction is an ORM invariant. A layer that also wrote would duplicate that or quietly work around it, which is why this is Queries and not a repository -- the smaller surface is the correct one, not a subset of one.

BoundFusion#

A declared fusion with a session behind it; call it to run it.

BoundFusion(fusion: Fusion, session: Any) -> None

BoundQuery#

A declared query with a session behind it; call it to run it.

BoundQuery(declaration: QueryDeclaration, session: Any) -> None

count#

async count(**values: Any) -> int

How many rows this query matches, without hydrating them.

Fusion#

Two or more declared searches, merged into one ranked answer.

Built by fuse, and used exactly as a declared query is -- an attribute of a Queries subclass, called with the parameters its searches need:

class Documents(Queries[Document]):
    nearest = (
        query()
        .order_by(Document.embedding.cosine_distance(Param("q")))
        .limit(50)
    )
    matching = (
        query(Document.search.matches(Param("terms")))
        .order_by(Document.search.rank(Param("terms")).desc())
        .limit(50)
    )
    hybrid = fuse(nearest, matching).limit(10)

found = await Documents(session).hybrid(q=vector, terms="llama husbandry")

The merge is by rank, not by score. A cosine distance of 0.18 and a ts_rank of 0.06 cannot be added, subtracted, or weighted against each other: they are different units, and any constant that reconciles them is wrong again the next time the embedding model changes. Reciprocal-rank fusion throws the numbers away and keeps the positions -- score = sum(1 / (k + rank)) over the searches that returned the row, ranks counted from 1. A row both searches placed mid-table beats a row one of them placed first, which is the behaviour hybrid retrieval is wanted for.

k damps how much a first place is worth; 60 is the conventional default from the paper the technique comes from, and lowering it makes the top of each search count for more.

Every search in a fusion is named, ordered, and bounded, and all three are checked when the class is defined. The rank is the position, so a query with no order_by has no ranking to contribute; and the bound is what keeps a fusion a merge of two shortlists rather than of two whole tables -- for a vector search it is also the LIMIT that lets the approximate index answer at all.

Named means fuse takes the attributes, never a query(...) written inside the call: a search on no class is in no declarations() listing, and the tools that walk a query set by name -- the transitional-column scanner among them -- would skip it in silence. A search declared on another Queries class is named, and fusing one of those is fine.

Ties in the fused score are broken by primary key ascending, so the same data returns the same order.

Fusion(halves: tuple[QueryDeclaration, ...], *, k: int, limit: int | None = None, settled: bool = False, name: str = 'fusion', parameters: tuple[str, ...] = ()) -> None

k (property)#

k: int

The reciprocal-rank constant this fusion scores with.

limit#

limit(value: int) -> Fusion

Cap the fused rows. Fixed at declaration time, as query()'s is.

limit_ (property)#

limit_: int | None

How many rows the fusion returns, or None for all of them.

Param#

One named value a declared query binds per call.

Param("paddock") stands where a literal would:

by_paddock = query(Llama.paddock_id == Param("paddock"))

It subclasses the column expression rather than the value expression for a reason worth knowing: Python gives the right operand of a comparison first refusal only when its type is a proper subclass of the left operand's, and the left operand here is a column. Inheriting from RelatedColumnExpr -- the deepest column type a declaration can compare against -- is what lets both Llama.paddock_id == Param(...) and Llama.paddock.name == Param(...) build a placeholder instead of trying to coerce one into a bigint. A Param is bait for an operator, never a node in a tree; the tree gets a Placeholder.

Because interception happens in the operator, a parameter works with the six comparisons (==, !=, <, <=, >, >=) in either order.

Operators spelled as methods -- like, the jsonb and array operators, the vector distances -- bind their operand before this class can intercept anything, so they reach it the other way round: _bind recognises the _as_placeholder seam and asks for a placeholder of the type that position takes. in_ is the exception, because its operand is a list of values rather than one, and the list's length is part of the query's shape.

Param(name: str) -> None