To start running queries, you need to bind a model to a database connection:

import sqlite3

Show.bind(sqlite3.connect('app.db'))

shows = (Show.query()
    .where('sold_out', '=', False)
    .orderBy('starts_at')
    .run()
)
# [Show(id=1, title='Nightcrawler', ...), ...]

Sustained works with any DB-API 2.0 connection passed to it; it will not open and manage connections itself.

Every statement runs parameterized: values travel as parameters and never as text inside the SQL.

The examples use the venue booking schema from Recipes, which extends the one in Getting Started with tickets and artists.

Matching the driver to the dialect

You can generally use any DB-API 2.0 driver to connect to your database backaend of choice, the only caveat is that connection’s parameter style has to match the placeholder Sustained uses for the dialect:

Dialect	Placeholder	Driver
DEFAULT, MSSQL, PRESTO, DUCKDB, ATHENA	?	sqlite3, pyodbc, trino, duckdb, pyathena with pyathena.paramstyle = "qmark"
POSTGRES, MYSQL	%s	psycopg, PyMySQL

SQL Dialects provides more information about each supported driver and gives sample connection patterns.

On Athena, run() also converts each parameter to the string form that the Athena API expects before it hands the parameters to pyathena, and it rewrites the placeholder of a None parameter as a literal NULL. If you execute to_sql() output yourself there, pass it through compiler.prepare_execution(sql, params) first.

Binding a connection

Model.bind() sets the connection as a class attribute which subclasses inherit:

from sustained import Model

Model.bind(conn)     # every model in the process
Show.bind(conn)      # only Show
Show.unbind()        # remove it again

A connection passed to run() or first() overrides any binding. This is useful when, for example, one query needs to run against a replica while the rest use the primary.

Sustained looks for a connection first as an argument that you passed to the call, then via a connection pinned by an open transaction() block, then the model’s binding, whether set on the model or inherited from a parent class. Running with none of those raises RuntimeError.

Reading rows

run() executes the query and hydrates each row into a model instance, using the column names from the cursor description:

for show in Show.query().where('venue_id', '=', 1).run():
    print(show.title, show.starts_at)

first() adds LIMIT 1 and returns one instance, or None when nothing matches. On MSSQL a query with no orderBy() and no offset() gets TOP 1 instead, because T-SQL writes LIMIT as OFFSET ... FETCH, which needs an ORDER BY. It leaves the original query alone, so a builder you keep around is still safe to reuse:

show = Show.query().where('title', '=', 'Nightcrawler').first()

Other result data structures

run() returns model instances by default. To get return results as another data structure, use one of these methods instead:

Method	Returns
to_dicts()	plain dicts keyed by column name
to_df()	a pandas DataFrame, keeping the column names even when empty
to_arrow()	a pyarrow Table

pandas and pyarrow are optional dependencies. The methods raise RuntimeError if the library is missing.

Type checking

The builder is generic over its model. Show.query() is a QueryBuilder[Show], and every chained clause retains this type parameter, so mypy and Pyright read the results without needing a cast or an annotation:

shows = (Show.query()
    .where('sold_out', '=', False)
    .orderBy('starts_at')
    .run()
)
# shows: List[Show]

show = Show.query().first()
# show: Optional[Show]

rows = Show.query().to_dicts()
# rows: List[Dict[str, Any]]

insert(), insert_from(), create_table_as(), update(), and delete() return a WriteBuilder[Show], whose run() is the affected row count, or the RETURNING rows as dicts:

removed = (Show.query()
    .delete()
    .where('starts_at', '<', cutoff)
    .run()
)
# removed: Union[int, List[Dict[str, Any]]]

QueryBuilder and WriteBuilder are one class at runtime. The two names exist so that a type checker can easily distinguish a row count from a list of models. isinstance(query, WriteBuilder) is true for any builder, so do not test against that comparison.

The columns are not typed. A select() does not narrow the model, and to_dicts() values stay RowValue, which is Any, because Python has no reasonable way to infer narrower types back out of the SQL string.

Untyped handoffs

Sustained exports type names for the connections and values that pass between your code and the library. You can import them from sustained:

from sustained import Binding, Connection, Cursor, RowValue, SqlValue

Connection and Cursor describe the DB-API 2.0 methods Sustained calls. They are protocols, so a sqlite3.Connection, a psycopg connection, and a pyodbc connection all match without subclassing. If you annotate a config module’s factory with Connection, a type checker will expect a real driver:

# sustained_config.py
import sqlite3

from sustained import Connection

def get_connection() -> Connection:
    return sqlite3.connect('app.db')

Model.bind() and every connection= argument expect a value of type Binding: typically via either a Connection or a ConnectionPool.

SqlValue and RowValue split values by direction. SqlValue is a value on its way into the database, whether bound as a parameter or rendered as a literal, and it is object rather than Any, so passing a query value where a column name belongs is still a type error. RowValue is a value read back, and it stays Any because the driver decides whether a NUMERIC arrives as a Decimal or a float.

Writing rows

insert(), update(), and delete() turn the builder into a write statement. They accept the same where() methods as a SELECT and use the same parameterized rendering:

Show.query().insert({'venue_id': 1, 'title': 'Nightcrawler'}).run()
# INSERT INTO shows (venue_id, title) VALUES (?, ?)

Show.query().update({'sold_out': True}).where('id', '=', 1).run()
# UPDATE shows SET sold_out = ? WHERE id = ?

Ticket.query().delete().where('sold_at', 'IS', None).run()
# DELETE FROM tickets WHERE sold_at IS NULL

A write renders only its where() clauses and returning(). Any other clause, such as orderBy(), limit(), or a join, raises ValueError when the statement renders, because the write would touch every matching row and not only the rows that clause selects. To cap a write, pick the rows in a subquery:

oldest = Ticket.query().select('id').orderBy('sold_at').limit(100)
Ticket.query().delete().whereIn('id', oldest).run()
# DELETE FROM tickets WHERE id IN (SELECT id FROM tickets ORDER BY sold_at ASC LIMIT 100)

MySQL refuses LIMIT inside an IN subquery. On MySQL, wrap the subquery in a second one with from_().

A write commits when it finishes, unless it is inside a transaction, and returns the affected row count. A write that raises outside a transaction rolls back before the error reaches you, so the rows that a batch insert sent before the failing row do not stay pending for the next commit. arun() does the same. Inside a transaction the rollback belongs to the block.

The count is -1 when the driver reports none, which asyncpg does for a batched multi-row insert and for any statement whose status string ends without a number. If you need an exact count, add returning(). The write then returns one row for each row it wrote, so len() of that list is the count.

A multi-row insert takes a list, and every row in it must have the same columns so that the statement has one template:

Ticket.query().insert([
    {'show_id': 1, 'price': 45},
    {'show_id': 1, 'price': 65},
]).run()

Without a RETURNING clause, a multi-row insert goes through the driver’s executemany() with a single-row template, which is the fast path for bulk loads.

Required WHERE for UPDATE and DELETE

An update() or delete() with no where() raises ValueError before it reaches the database.

To write every row deliberately, use a predicate that is always true:

from sustained.builder import QueryBuilder

(Show.query()
    .update({'sold_out': False})
    .where(QueryBuilder.raw('1'), '=', 1)
    .run()
)

Upserts

Chain onConflict(columns) after insert(), then merge() to update the existing row or ignore() to leave it:

(Artist.query()
    .insert({'name': 'Low', 'country': 'US'})
    .onConflict('name')
    .merge()
    .run()
)

merge() updates every inserted column except the conflict columns, or only the columns in a list you pass it. The conflict columns need a unique constraint or primary key in the database, or the engine rejects the statement. If every inserted column is also a conflict column, merge() has nothing left to update and raises ValueError.

Postgres, SQLite, and DuckDB render ON CONFLICT, MySQL renders ON DUPLICATE KEY UPDATE, MSSQL renders a MERGE statement, Athena renders MERGE for Iceberg tables, and Presto raises DialectError.

RETURNING

returning() adds a RETURNING clause on dialects that support it, and the statement then returns a list of dicts instead of a row count:

rows = (Show.query()
    .insert({'venue_id': 1, 'title': 'Nightcrawler'})
    .returning('id')
    .run()
)
# [{'id': 42}]

MySQL, MSSQL, Presto, and Athena raise DialectError. On MSSQL, use an OUTPUT clause through raw SQL instead, and on MySQL, read the row back with a second query.

INSERT … SELECT and CREATE TABLE AS

insert_from(columns, query) inserts another query’s result. create_table_as(name, temporary=False) turns a SELECT into a CTAS statement:

class ShowArchive(Model):
    tableName = 'show_archive'

past = Show.query().select('id', 'title').where('starts_at', '<', '2026-01-01')

ShowArchive.query().insert_from(['id', 'title'], past).run()
# INSERT INTO show_archive (id, title)
# SELECT id, title FROM shows WHERE starts_at < ?

(Show.query()
    .select('id')
    .where('sold_out', '=', True)
    .create_table_as('sellouts')
    .run()
)
# CREATE TABLE sellouts AS SELECT id FROM shows WHERE sold_out = ?

insert_from() writes to the table of the model in front of .query() and reads from the query you pass in.

MSSQL raises DialectError for CTAS, so use SELECT INTO through raw SQL there.

Transactions

Model.transaction() opens a block that commits at the end and rolls back on an exception. Statements inside share one transaction, and run() stops committing per statement:

with Show.transaction():
    Show.query().update({'sold_out': True}).where('id', '=', 1).run()

    (Ticket.query()
        .delete()
        .where('show_id', '=', 1)
        .andWhere('sold_at', 'IS', None)
        .run()
    )

Nested blocks use savepoints, so a failure inside an inner block rolls back only that block and the outer transaction continues. The savepoint statement follows the model’s dialect: MSSQL gets SAVE TRANSACTION, everything else the ANSI SAVEPOINT. DuckDB has no savepoints, so a nested block raises DialectError there.

A connection that you put in autocommit, such as sqlite3.connect(autocommit=True) or a psycopg connection with autocommit on, commits every statement as it runs and ignores rollback(). The block reads the connection’s autocommit attribute and, when it is True, sends BEGIN, COMMIT, and ROLLBACK as statements, so the block still rolls back.

Eager loading relations

withGraphFetched() loads the relations named in relationMappings when the query runs, at one extra query per relation:

venues = Venue.query().withGraphFetched('shows').run()

for venue in venues:
    for show in venue.shows:
        print(venue.name, show.title)

A HasManyRelation or ManyToManyRelation attaches a list, and the to-one relation types attach a single instance or None.

Eager loading matches rows on the join key, so both result sets need that column, and you must either keep it in your select() or select every column. A relation through a link table loads with one query that joins the link table to the far table.

A dotted path loads a relation of a relation:

venues = Venue.query().withGraphFetched('shows.tickets').run()

for venue in venues:
    for show in venue.shows:
        print(show.title, len(show.tickets))

Each level costs one query, batched over every instance at that level, so three venues with twelve shows load their tickets with one query, not twelve. Paths that share a prefix load the prefix once, so withGraphFetched('shows.tickets', 'shows.artists') runs three queries after the first. An unknown segment raises ValueError when you build the query, naming the segment and the model it was read from.

A join flattens the related rows into the result and repeats the parent row once per child. Eager loading costs one extra query and returns each object separately; a join costs one query and gives you a wide, flattened result.

Connection pooling

ConnectionPool creates connections from a factory, up to max_size, and reuses released ones. You bind it exactly like a connection:

from sustained.pool import ConnectionPool

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

shows = Show.query().where('sold_out', '=', False).run()

Each statement checks a connection out for its duration and releases it after. A transaction pins one connection to the thread for the whole block, so the savepoints and the commit land on the same session:

with Show.transaction():
    Show.query().update({'sold_out': True}).where('id', '=', 1).run()
    Ticket.query().insert({'show_id': 1, 'price': 45}).run()

If you pass the pool to a query by hand inside the block, the query also runs on the pinned connection, so query.run(pool) stays in the transaction instead of checking out a second connection.

An exhausted pool raises PoolTimeout after the configured timeout rather than blocking forever. A waiting caller also wakes when the pool drops a broken connection, and it opens a new connection in the freed slot. pool.close() closes the idle connections, and a caller still waiting gets RuntimeError.

A transaction belongs to the thread that opened it. A second thread that calls transaction() on the same connection gets a RuntimeError, because a connection runs one transaction at a time and the block would otherwise nest inside the first thread’s. Give each thread its own connection, from a pool or from a second bind target.

Async execution

Async queries run through an adapter, which is the async equivalent of a connection:

Adapter	Wraps
AsyncpgAdapter	asyncpg, converting %s placeholders to $1..$n
AiosqliteAdapter	aiosqlite
DbApiAsyncAdapter	any synchronous DB-API connection, in a worker thread

Bind one with bind_async(), then use the a-prefixed methods:

from sustained.aio import DbApiAsyncAdapter

Show.bind_async(DbApiAsyncAdapter(sqlite3.connect('app.db', check_same_thread=False)))

shows = await Show.query().where('sold_out', '=', False).arun()
show = await Show.query().where('id', '=', 1).afirst()
rows = await Show.query().ato_dicts()

async with Show.async_transaction():
    await Show.query().update({'sold_out': True}).where('id', '=', 1).arun()

arun() mirrors run(), including hydration, RETURNING rows, batched multi-row inserts, and eager loading. Both paths share the same loader, so dotted paths, per-level batching, and relations through a link table behave the same way. async_transaction() mirrors transaction() as well: nested blocks open a savepoint, so an inner failure rolls back only the inner block. A rolled-back savepoint is released afterward, so the same block can run again under the same name.

async_transaction() opens and closes the block the way the driver expects. DbApiAsyncAdapter wraps a DB-API 2.0 driver, which opens the transaction itself, so the block sends no BEGIN and ends with the driver’s commit() or rollback(). The exception is a connection that you put in autocommit mode, which commits every statement as it runs, so its blocks send the transaction statements instead. AsyncpgAdapter runs in autocommit and treats commit() as a no-op, so its blocks send BEGIN, COMMIT, and ROLLBACK as statements. A dialect whose driver has no transaction control, such as DuckDB, also gets the statements. A custom adapter declares which kind it is by overriding driver_transaction_control(), which returns False on the base class.

Async pooling

One adapter runs one statement at a time, because a connection runs one transaction at a time, DbApiAsyncAdapter serializes every call behind one lock, and asyncpg sends one statement per connection. Ten concurrent arun() calls on one adapter therefore queue up behind each other.

AsyncConnectionPool is the async twin of ConnectionPool. It opens adapters from an async factory, up to max_size, and you bind it the same way:

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

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

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

shows, tickets = await asyncio.gather(
    Show.query().arun(),
    Ticket.query().arun(),
)

Each call checks one adapter out for its whole length, including the statement, its eager loads, and its commit, then gives it back. An async_transaction() block keeps one adapter from BEGIN to COMMIT. If you pass the same pool to a call inside that block, such as arun(query, pool), the call runs on the adapter that the block checked out, and a nested async_transaction(pool) opens a savepoint on it, the same way the blocking transaction(pool) nests. The pool rolls back each released adapter first, so a failed statement never reaches the next task. If an adapter’s driver refuses a rollback with no transaction open, as duckdb does, the pool instead probes the adapter with SELECT 1 and keeps it when the probe succeeds. A task cancelled while it gives an adapter back gets its CancelledError at once, and the rollback finishes in the background, so the adapter still returns to the pool. An exhausted pool raises PoolTimeout, the same error the blocking pool raises, and a waiting task wakes for a freed slot as it does there. The factory runs outside the pool’s bookkeeping, so a slow connect does not hold up a release. await pool.close() closes the idle adapters.

The pool runs no statements itself, so await pool.fetch(...) raises, because a write and its commit would otherwise land on two different connections. When you want to run something by hand, take an adapter out with async with pool.scope() as adapter. AsyncMigrator takes an adapter rather than a pool, because a migration run needs to stay on one session.

Watching what runs

set_statement_listener() registers a function called after every executed statement, with the SQL text, the parameter tuple, and the duration in seconds:

from sustained.execution import set_statement_listener

def log(sql, params, seconds):
    if seconds > 0.1:
        print(f'{seconds:.3f}s  {sql}')

set_statement_listener(log)
set_statement_listener(None)   # remove it

Only one listener can be registered at a time, and it sees every statement in the process, including the migrator’s.

Where to go next

You want to	Read
Pick a driver for your engine	SQL Dialects
Create the tables you are querying	Schema and Migrations
Find the code for a specific task	Recipes
See every method and what it raises	Execution reference