sustained.execution, sustained.pool, sustained.aio, and sustained.rendering: everything between a finished query and the database.

Guide: Executing Queries.

Connection resolution

Every execution method resolves a connection in the same order:

1. The connection argument, when you pass one.
2. A connection pinned to this thread by an open transaction() block.
3. The binding from Model.bind().

When none of the three resolves, the call raises RuntimeError. A ConnectionPool found at step 1 or step 3 is checked out for the length of the statement and released afterward.

Connection types

These live in sustained.types and are re-exported from sustained.

Name	What it is
Connection	Protocol: cursor(), commit(), rollback(), close().
Cursor	Protocol: execute(), executemany(), fetchone(), fetchall(), close(), plus description and rowcount.
Binding	Union[Connection, ConnectionPool], the type bind() and every connection= argument take.
SqlValue	A value going into the database. Alias for object.
RowValue	A value read back out. Alias for Any.

The two protocols list only the methods Sustained calls, so a driver connection matches by having those methods. No class subclasses the protocols, and no class registers with them.

Transactions

These live in sustained.execution. Use them through Model.transaction() rather than calling them yourself.

transaction(connection, dialect=None)

A context manager. Commits when the block finishes, and rolls back on any exception. The dialect chooses the savepoint spelling for nested blocks; Model.transaction() passes the model’s dialect for you.

Statements inside the block share one cursor. DuckDB requires this, because its driver autocommits every statement and gives every cursor its own session, so the block opens, commits, and rolls back the transaction with SQL on that shared cursor.

in_transaction(connection) -> bool

Whether a transaction is open on this connection.

connection_scope(explicit, binding)

The resolution above, as a context manager.

Nested blocks on one connection use savepoints named sustained_sp_<depth>, so a failure in an inner block rolls back only that block. The statement follows the dialect: ANSI SAVEPOINT everywhere except MSSQL, which gets SAVE TRANSACTION. DuckDB has no savepoints, so a nested block raises DialectError before any statement runs. Inside a transaction, run() stops committing per statement.

With a pool, the first block checks a connection out and pins it to the thread. Every statement in the block uses that one connection, and nested blocks reuse it as savepoints.

Statement logging

set_statement_listener(listener)

Registers a callable that runs after every executed statement, with the SQL text, the parameter tuple, and the duration in seconds. Pass None to remove the listener.

The listener is global: one listener covers every model, every connection, and every thread.

from sustained.execution import set_statement_listener

set_statement_listener(lambda sql, params, seconds: log.info('%s %r %.3fs', sql, params, seconds))

Hydration and eager loading

fetch_models(model_class, cursor) -> list[Model]

Builds instances from a cursor. Returns [] when the cursor has no description.

eager_load_relation(model_class, connection, parents, relation_name)

Runs one query and attaches the results to the parent instances. Raises ValueError for an unknown relation name, for a join reference that is not table.column, and for rows that are missing the join column. A HasManyRelation attaches a list, and the to-one relation types attach a single instance or None.

ConnectionPool

ConnectionPool lives in sustained.pool.

ConnectionPool(factory, max_size=5, timeout=30.0)

The pool creates connections from factory as it needs them, up to max_size, and reuses released connections. Bind a pool the way you would bind a connection. A max_size below 1 raises ValueError.

size -> int

Property. The number of connections created so far.

connection()

A context manager that checks a connection out and releases it at the end of the block.

acquire_raw()

Checks a connection out. You have to release it yourself. Raises PoolTimeout when the pool stays exhausted past timeout, and RuntimeError when the pool is closed. A factory that raises does not consume a slot.

release(connection)

Returns the connection to the pool, or closes it when the pool is closed. Any open transaction is rolled back first, on every release, so the next caller never inherits a stale snapshot or an aborted transaction. When the rollback raises, the pool probes the connection with SELECT 1: one that answers is kept, because some drivers, duckdb among them, refuse rollback with no transaction open rather than reporting a broken connection. One that does not answer is closed and dropped. A connection the pool did not hand out raises ValueError, which catches a double release.

close()

Closes the idle connections. A checked-out connection closes on release.

from sustained.pool import ConnectionPool

pool = ConnectionPool(lambda: psycopg.connect(DSN), max_size=10)
Model.bind(pool)

Async adapters

These live in sustained.aio. Every adapter has the same methods, so a query does not need to know which driver is underneath.

Method	Returns
await fetch(sql, params)	(column_names, rows)
await execute(sql, params)	affected row count, or -1
await executemany(sql, seq_of_params)	affected row count, or -1
await commit()	None
await rollback()	None
await close()	None
async with scope()	the adapter one call runs on
driver_transaction_control()	whether the driver opens the transaction
await begin_where_ddl_autocommits()	the BEGIN such a driver still needs

A row count of -1 means the driver reported none. Add returning() to the write when you need an exact count.

driver_transaction_control() tells async_transaction() how to open and close a block. It returns False on the base class and on AsyncpgAdapter, so the block runs BEGIN, COMMIT, and ROLLBACK as statements. DbApiAsyncAdapter returns True, because a DB-API 2.0 driver opens the transaction itself; the block then ends with commit() or rollback(). It returns False when the connection it wraps reports autocommit as True, because such a connection commits every statement as it runs and its commit() closes nothing. begin_where_ddl_autocommits() covers the one gap in that promise: sqlite3 in legacy transaction control leaves schema statements outside its implicit transaction, so DbApiAsyncAdapter sends a BEGIN there.

Every call opens scope() before it runs. A plain adapter yields itself; a pool yields one of its adapters and takes it back at the end, so a statement and its commit stay on one connection.

DbApiAsyncAdapter(connection)

Wraps any synchronous DB-API connection. Runs each call in a worker thread under a lock. The connection has to permit cross-thread use, as in sqlite3.connect(..., check_same_thread=False).

AiosqliteAdapter(connection)

Wraps aiosqlite and awaits the driver directly.

AsyncpgAdapter(connection)

Wraps asyncpg. Converts %s placeholders to $1..$n. asyncpg is autocommit, so commit() and rollback() do nothing, driver_transaction_control() is False, and executemany() returns -1. execute() reads its count out of the status string and returns -1 when the status reports none.

AsyncAdapter is the abstract base class. Subclass it for a driver that has no adapter here. close() does nothing on the base, for an adapter that borrows a connection it does not own.

AsyncConnectionPool

AsyncConnectionPool lives in sustained.aio_pool.

AsyncConnectionPool(factory, max_size=5, timeout=30.0)

Pools adapters from an async factory, up to max_size, and reuses released ones. Bind it the way you bind an adapter. A max_size below 1 raises ValueError. One adapter runs one statement at a time, so a pool is how concurrent async queries reach the database in parallel.

The pool is an AsyncAdapter so it can be bound and passed like one, but it runs no statement itself: fetch(), execute(), executemany(), commit(), and rollback() all raise RuntimeError, because a write and its commit would land on two different connections.

size -> int

Property. The number of adapters opened so far.

async with scope()

Checks an adapter out and releases it at the end of the block.

await acquire()

Checks an adapter out. You have to release it yourself. Raises PoolTimeout when the pool stays exhausted past timeout, and RuntimeError when the pool is closed.

await release(adapter)

Gives an adapter back, rolling it back first so a failed statement does not reach the next task. An adapter the pool did not hand out raises ValueError, which catches a double release. When the rollback raises, the pool probes the adapter with SELECT 1: one that answers is kept, because some drivers, duckdb among them, refuse rollback with no transaction open rather than reporting a broken connection. One that does not answer is closed and dropped, and its slot reopens.

await close()

Closes the idle adapters and refuses new checkouts. A checked-out adapter closes on release.

from sustained.aio import AsyncpgAdapter
from sustained.aio_pool import AsyncConnectionPool

async def open_adapter():
    return AsyncpgAdapter(await asyncpg.connect(DSN))

Model.bind_async(AsyncConnectionPool(open_adapter, max_size=10))

AsyncMigrator takes an adapter rather than a pool, because a migration run belongs on one session.

Async helpers

async_transaction(adapter, dialect=None)

An async context manager that opens, commits, and rolls back the transaction. An adapter whose driver has no transaction control gets the dialect’s own statements; the default dialect issues BEGIN, COMMIT, and ROLLBACK. An adapter over a DB-API 2.0 driver ends the block with commit() or rollback() instead, because that driver opened the transaction itself. Nested blocks on one adapter use savepoints named sustained_sp_<depth>, spelled per dialect; a dialect with no savepoints raises DialectError on nesting. A failed inner block rolls back to its savepoint and then releases it, so a later block can take the same name. If the rollback itself fails, the block’s own error still propagates, with the rollback failure as its cause. Model.async_transaction() passes the model’s dialect. Sustained tracks the nesting per adapter, so give each concurrent task its own adapter.

Handed a pool, the block checks one adapter out and keeps it to the end. Inside the block, a call handed the same pool runs on that adapter, and a nested async_transaction(pool) opens a savepoint on it instead of checking a second adapter out, which would commit on its own and deadlock a pool of one.

in_async_transaction(adapter) -> bool

Whether a transaction is open on this adapter.

resolve_adapter(explicit, model_class)

Resolves the argument, then the open transaction, then Model.bind_async(). Raises RuntimeError when none of the three resolves.

await run_async(query, adapter=None)

The function arun() calls.

convert_format_to_numbered(sql) -> str

Rewrites %s markers to $1..$n.

The transaction pin lives in a ContextVar, so it follows the task tree rather than the thread.

Async eager loading shares the planner the synchronous path uses, so it covers dotted paths and through relations.

Rendering

sustained.rendering is internal, and visible when a custom expression renders itself.

RenderContext(compiler, parameterize=False) contains the compiler and the value-handling mode. ctx.value(v) returns a placeholder and collects the value when parameterize is set, and returns a formatted SQL literal when it is not. An Expression renders as written in either mode. Both str(query) and to_sql() run through this one code path.

bind_raw(sql, params, ctx) fills the ? markers in a raw fragment, and raises ValueError when the marker count does not match the parameter count.