Sustained.py

A Python ORM, SQL query builder, and schema migration tool, inspired by Objection.js and Knex

Start here

Recipes

Each recipe is a standalone snippet that completes one common task.

Every example uses the venue booking schema from Getting Started, extended with artists, tickets, and a link table for the join examples.

from sustained import Model, RelationType
from sustained.schema import Boolean, Integer, Numeric, String, Timestamp


class Venue(Model):
    tableName = 'venues'
    tableColumns = {
        'id': Integer(primary_key=True, autoincrement=True),
        'name': String(120, nullable=False),
        'city': String(80, nullable=False),
        'capacity': Integer(),
    }
    relationMappings = {
        'shows': {
            'relation': RelationType.HasManyRelation,
            'modelClass': 'Show',
            'join': {'from': 'venues.id', 'to': 'shows.venue_id'},
        },
    }


class Artist(Model):
    tableName = 'artists'
    tableColumns = {
        'id': Integer(primary_key=True, autoincrement=True),
        'name': String(120, unique=True, nullable=False),
        'country': String(2),
    }
    relationMappings = {
        'shows': {
            'relation': RelationType.ManyToManyRelation,
            'modelClass': 'Show',
            'join': {
                'from': 'artists.id',
                'through': {
                    'from': {'table': 'show_artists', 'key': 'artist_id'},
                    'to': {'table': 'show_artists', 'key': 'show_id'},
                },
                'to': 'shows.id',
            },
        },
    }


class Show(Model):
    tableName = 'shows'
    tableColumns = {
        'id': Integer(primary_key=True, autoincrement=True),
        'venue_id': Integer(references='venues.id', nullable=False),
        'title': String(200, nullable=False),
        'starts_at': Timestamp(),
        'sold_out': Boolean(default=False),
    }
    relationMappings = {
        'venue': {
            'relation': RelationType.BelongsToOneRelation,
            'modelClass': 'Venue',
            'join': {'from': 'shows.venue_id', 'to': 'venues.id'},
        },
        'tickets': {
            'relation': RelationType.HasManyRelation,
            'modelClass': 'Ticket',
            'join': {'from': 'shows.id', 'to': 'tickets.show_id'},
        },
        'artists': {
            'relation': RelationType.ManyToManyRelation,
            'modelClass': 'Artist',
            'join': {
                'from': 'shows.id',
                'through': {
                    'from': {'table': 'show_artists', 'key': 'show_id'},
                    'to': {'table': 'show_artists', 'key': 'artist_id'},
                },
                'to': 'artists.id',
            },
        },
    }


class Ticket(Model):
    tableName = 'tickets'
    tableColumns = {
        'id': Integer(primary_key=True, autoincrement=True),
        'show_id': Integer(references='shows.id', nullable=False),
        'price': Numeric(10, 2),
        'sold_at': Timestamp(),
    }

Querying

Reuse one filter for several queries

A builder changes in place, so a shared base query would collect the filters from every branch. To keep the branches apart, give each one a deep copy of the base query with clone().

sold_out = Show.query().where('sold_out', '=', True)

fillmore = sold_out.clone().where('venue_id', '=', 1).run()
first_ave = sold_out.clone().where('venue_id', '=', 2).run()

Without clone(), the second line would filter on both venues at once and return nothing.

Filter with Python operators instead of strings

Model.c gives every column a typed reference. Comparison operators build a condition, and &, |, and ~ combine conditions.

from sustained import col

Show.query().where((Show.c.sold_out == True) & ~(Show.c.title.like('Cancelled%')))

# For a table with no model in scope:
Show.query().innerJoinRelated('venue').where(col('venues.capacity') > 1400)

Use & and | rather than and and or. A Predicate raises TypeError in a boolean context, so the keyword version fails with an error instead of evaluating to one side.

Group a set of OR conditions

You can pass a lambda to any where method. The lambda receives a builder whose conditions render inside parentheses, which lets you group and nest clauses.

Show.query().where('venue_id', '=', 1).andWhere(
    lambda q: q.where('sold_out', '=', True).orWhere('starts_at', '>', '2026-09-01')
)
# WHERE venue_id = 1 AND (sold_out = TRUE OR starts_at > '2026-09-01')

Groups can nest arbitrarily deep. The only limit is what your peers are willing to code review. The first condition in each chain must be a plain where rather than andWhere or orWhere.

Filter the result of an aggregate

having() takes the aggregate call itself, such as 'SUM(tickets.price)', because standard SQL does not let HAVING refer to column aliases.

revenue = (
    Ticket.query()
    .select('venues.name')
    .sum('tickets.price', alias='revenue')
    .count('tickets.id', alias='sold')
    .innerJoin('shows', 'tickets.show_id', '=', 'shows.id')
    .innerJoin('venues', 'shows.venue_id', '=', 'venues.id')
    .groupBy('venues.name')
    .having('SUM(tickets.price)', '>', 50)
    .orderBy('venues.name')
)

revenue.to_dicts()
# [{'name': 'First Avenue', 'revenue': 60, 'sold': 1},
#  {'name': 'The Fillmore', 'revenue': 120, 'sold': 3}]

Rank rows inside a group

ranked = (
    Ticket.query()
    .select('show_id', 'price')
    .select_window(
        'ROW_NUMBER', 'rank',
        partition_by=['show_id'],
        order_by=['price DESC'],
    )
)
# SELECT show_id, price,
#        ROW_NUMBER() OVER (PARTITION BY show_id ORDER BY price DESC) AS rank
# FROM tickets

To filter on rank, wrap this in a subquery. On DuckDB, the only supported dialect with QUALIFY, you can use qualify('rank <= 3') instead.

Label rows with CASE

Show.query().select('title').select_case(
    'status', 'open', when_clauses=[('sold_out = 1', 'sold out')]
)
# SELECT title, CASE WHEN sold_out = 1 THEN 'sold out' ELSE 'open' END AS status

select_case() treats each string result as a value, so wrap a result in col('other_col') when it names a column.

Feed one query into another with a CTE

sold_out = Show.query().select('venue_id').where('sold_out', '=', True)

Venue.query() \
    .with_('sold_out_shows', sold_out) \
    .select('venues.name') \
    .innerJoin('sold_out_shows', 'venues.id', '=', 'sold_out_shows.venue_id')

A raw join needs a table name, so you turn a derived result set into a CTE first and then join it by its alias. Pass recursive=True to with_() for a self-referencing CTE.

from sustained import QueryBuilder

Venue.query().select('name').whereExists(
    Show.query()
    .select(QueryBuilder.raw('1'))
    .where('shows.venue_id', '=', QueryBuilder.raw('venues.id'))
    .andWhere('sold_out', '=', True)
)

Wrap the outer query’s column in QueryBuilder.raw(), because without it Sustained binds 'venues.id' as a string value instead of comparing against the column.

Page through a large table

page() uses LIMIT and OFFSET, which is fine for early pages and gets slower as the offset grows, because the database still walks the skipped rows.

Show.query().page(2, 25)   # zero-based page number: rows 51-75

cursor_page() filters on the last value seen instead, so the cost stays flat however deep you go.

first = Show.query().cursor_page('id', 100).run()
second = Show.query().cursor_page('id', 100, after=first[-1].id).run()

The cursor column must be unique and sorted, or the next page skips rows that share the last row’s value.

Count the matches without running the query

total() runs SELECT COUNT(*) over the query with ordering, LIMIT, and OFFSET stripped, and leaves the original builder alone.

query = Show.query().where('sold_out', '=', True)

count = query.total()   # 2
rows = query.limit(20).run()

Get rows as dicts, a DataFrame, or Arrow table

Show.query().to_dicts()   # [{'id': 1, 'title': 'Opening Night', ...}]
Show.query().to_df()      # pandas DataFrame
Show.query().to_arrow()   # pyarrow Table

pandas and pyarrow are optional dependencies, and when one is not installed, the method that needs it raises a RuntimeError naming the missing package.

Relations

Load a relation with the parent rows

for venue in Venue.query().withGraphFetched('shows').run():
    print(venue.name, [s.title for s in venue.shows])
# The Fillmore ['Opening Night', 'Second Night']
# First Avenue ['Tour Finale']

Each relation costs one extra query. A HasManyRelation attaches a list; the to-one types attach a single instance or None. Keep the join key columns in your select(), or the second query will have nothing to match on.

Join a many-to-many relation

Declare the link table in the through mapping, then join by relation name. Sustained joins the link table for you.

Artist.query().select('artists.name', 'shows.title').innerJoinRelated('shows')
# SELECT artists.name, shows.title
# FROM artists
# INNER JOIN show_artists ON artists.id = show_artists.artist_id
# INNER JOIN shows ON show_artists.show_id = shows.id

The join type you name applies to the second hop. A left or full join makes the first hop a LEFT JOIN, so an artist with no link row stays in the result. Every other join type makes the first hop an INNER JOIN.

Join the same table twice

Give the second join an alias and refer to it by that name. For a many-to-many relation, Sustained also renders the second copy of the link table as <alias>_<link table>, here support_show_artists.

Show.query() \
    .select('shows.title', 'headline.name', 'support.name') \
    .innerJoinRelated('artists', alias='headline') \
    .leftJoinRelated('artists', alias='support')

Reference a model defined in another file

Give modelClass the class name as a string. Every model with a tableName registers itself when its class is defined.

'modelClass': 'Show'

You must import the class before you build the query, or the name does not resolve and Sustained raises ValueError.

Writing data

Insert many rows at once

Ticket.query().insert([
    {'show_id': 1, 'price': 45.0},
    {'show_id': 1, 'price': 45.0},
    {'show_id': 2, 'price': 30.0},
]).run()

Every row must have the same keys. Without a RETURNING clause, this runs through the driver’s executemany(), which is the fast path for bulk loads.

Insert or update on conflict

Artist.query().insert({'name': 'Low', 'country': 'GB'}) \
    .onConflict('name').merge().run()      # update the existing row

Artist.query().insert({'name': 'Duster', 'country': 'US'}) \
    .onConflict('name').ignore().run()     # keep the existing row

The conflict columns must have a unique constraint or primary key in the database, or the statement fails there. merge() updates every inserted column except the conflict columns, or only the columns in a list you pass it. Postgres, SQLite, and DuckDB render ON CONFLICT, MySQL renders ON DUPLICATE KEY UPDATE, MSSQL and Athena (on Iceberg tables) render MERGE, and Presto raises DialectError.

Get the generated id back from an insert

rows = Venue.query().insert({'name': 'Roseland', 'city': 'Portland'}) \
    .returning('id').run()
# [{'id': 3}]

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

Copy rows between tables

class TicketArchive(Model):
    tableName = 'ticket_archive'


recent = Ticket.query().select('show_id', 'price').where('sold_at', '>', '2026-01-01')

TicketArchive.query().insert_from(['show_id', 'price'], recent).run()

To copy into a new table instead of an existing one, create_table_as('name') turns a SELECT into a CTAS statement. MSSQL raises DialectError, so use SELECT ... INTO through raw SQL there.

Update every row on purpose

update() and delete() raise ValueError without a where(). To override this guardrail, use a WHERE clause that always evaluates to TRUE:

from sustained import QueryBuilder

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

Roll back a group of writes together

with Model.transaction():
    Show.query().insert({'venue_id': 1, 'title': 'Added Date'}).run()
    Ticket.query().insert({'show_id': 4, 'price': 40.0}).run()

The block commits when it finishes and rolls back on any exception. Inside a transaction, run() stops committing per statement. Nested blocks use savepoints, so an inner failure rolls back only the inner block. DuckDB has no savepoints, so a nested block raises DialectError there.

Schema changes

Add a column

Add the new column to tableColumns, then let Sustained work out the difference.

Show.tableColumns['support_act'] = String(200)

migrator.up(models=[Venue, Artist, Show, Ticket])
# ['auto_20260816133122_029439']

Pass every model you manage, not only the changed one, because Sustained does not keep a table up to date when its model is missing from the list.

Preview the migration before it runs

migration = migrator.plan([Venue, Artist, Show, Ticket])
if migration is not None:
    print(migration.up)
    # ['ALTER TABLE shows ADD COLUMN support_act VARCHAR(200)']

plan() returns None when the schema is current. It records nothing and applies nothing, so the migration it returns is not left pending. To test the migration on the real database before you apply it, pass the same models to rehearse(models=[...]). This works only on engines that can roll back schema changes, so on other engines, apply your changes to a test database instead.

If you want the differences without the SQL, diff_schema() reports every gap:

from sustained.autogenerate import diff_schema

print(diff_schema(conn, [Venue, Artist, Show, Ticket]).summary())
# add column shows.support_act
# drop column shows.legacy (destructive)

Rename a column instead of dropping and re-adding it

A database catalog cannot tell a rename from a drop plus an add, so pass an explicit hint when you rename:

migrator.up(models=models, renames={'shows.name': 'title'})
migrator.up(models=models, table_renames={'gigs': 'shows'})

Without the hint, the old column is left alone and the new one is added beside it, because the catalog gives no reason to believe the rows should move.

Drop a column

migrator.up(models=models, allow_drops=True)

A migration that contains a drop has no down step, because the dropped data cannot come back.

Add a NOT NULL column to a table with rows

Give the column a value for the rows that already exist, either a default or a backfill:

'country': String(2, nullable=False, backfill='US')

Sustained then generates three steps: add the new column as nullable, UPDATE the existing rows, then set NOT NULL. Without a value, generation refuses, because the existing rows would have no value to store.

Write a migration by hand

Live your best life fam, we won’t judge.

from sustained.migrations import Migration, Migrator

migrations = [
    Migration(
        'split_artist_name',
        up=[
            'ALTER TABLE artists ADD COLUMN first_name TEXT',
            "UPDATE artists SET first_name = substr(name, 1, instr(name, ' ') - 1) WHERE 1 = 1",
        ],
        down='ALTER TABLE artists DROP COLUMN first_name',
    ),
]

Migrator(conn, migrations).up()

A step is a SQL string, a list of statements, or a callable that receives the connection.

Keep a view in step with its definition

You replace a view rather than alter it step by step, so it fits badly in a versioned migration. Instead, put it in an <id>.repeat.sql file, which re-runs whenever its contents change.

-- upcoming_shows.repeat.sql
DROP VIEW IF EXISTS upcoming_shows;
CREATE VIEW upcoming_shows AS
  SELECT shows.title, venues.name AS venue
  FROM shows JOIN venues ON shows.venue_id = venues.id
  WHERE shows.sold_out = FALSE;

Keep the SQL safe to re-run by using CREATE OR REPLACE or by dropping the view first. Repeatables run after every versioned migration, have no down step, and down never touches them.

Fill a value into a SQL migration at load time

-- 003_grant.up.sql
GRANT SELECT ON shows TO ${reader};

from sustained.migration_files import load_migrations

migrations = load_migrations('migrations', placeholders={'reader': 'app_ro'})

Passing a mapping, even an empty one, turns substitution on: a ${key} with no value then raises ValueError naming the file and the key. $${ escapes a literal ${. Sustained substitutes values before it computes the checksum, so changing a value after a versioned migration ran shows up as a checksum mismatch.

Adopt a database that already has the schema

migrator.baseline('002_create_shows')   # record as applied, run nothing
migrator.up()                           # apply only what comes after

The baseline rows record checksums, so validation still catches later edits, and their null execution time marks them as never having run.

Deploys

Wire up the command line

The sustained command reads a config module, sustained_config by default:

# sustained_config.py
import psycopg

from models import Artist, Show, Ticket, Venue


def get_connection():
    return psycopg.connect('postgresql://localhost/app')


migrations_dir = 'migrations'
models = [Venue, Artist, Show, Ticket]
dialect = 'postgres'

Point at another module with --config mymodule. Naming models lets plan report model drift as well as pending migrations.

Gate a deploy on what is waiting

plan exits 0 when the database is current, 2 when work is waiting, and 1 when validation found problems.

sustained plan
case $? in
  0) echo "nothing to do" ;;
  2) sustained rehearse && sustained migrate ;;
  *) exit 1 ;;
esac

argparse also exits 2 on a usage error, so a script that treats 2 as “work is waiting” should check stderr for an error: line.

Read the plan in a pipeline

status, validate, plan, and rehearse all accept a --json flag.

$ sustained plan --json
{
  "pending": [],
  "problems": [],
  "drift": [
    {
      "sql": "ALTER TABLE \"shows\" ADD COLUMN \"support_act\" VARCHAR(200)",
      "destructive": false,
      "guards": []
    }
  ],
  "error": null
}

drift is null, not [], when the config module names no models, so a caller can tell “nothing was compared” from “compared and found no gap”. Each statement is an object with its SQL, whether it removes data, and the verdicts of any guards that flagged it. A pending migration lists its statements the same way under statements, which is null for a callable step because a callable renders no SQL. error is null unless the command failed.

Rehearse where the rollback cannot be trusted

Only SQLite, Postgres, and DuckDB roll schema changes back, so only those engines can rehearse against the real database. On other engines, point the rehearsal at a throwaway database:

def get_rehearsal_connection():
    return psycopg.connect('postgresql://localhost/app_rehearsal')

The scratch database is usually empty, so the whole history replays instead of only what is pending, which proves that the migrations run on an empty database. The changes may remain after the rollback there, so recreate the database before the next rehearsal. In Python, this is migrator.rehearse(scratch=True).

Hand the SQL to a DBA instead of running it

Because not everything needs to be your problem.

print(migrator.script('up'))     # every statement a run would execute
print(migrator.script('down'))   # the rollback

The output includes the tracking-table bookkeeping, so a run applied by hand still leaves the history correct.

Run something before and after a migration

The config module can define before_migrate, after_migrate, and on_error, and migrate calls whichever ones it finds:

def before_migrate(connection):
    notify('migration starting')


def after_migrate(connection, applied):
    notify(f'applied {len(applied)} migrations')


def on_error(connection, migration_id, error):
    page_someone(f'{migration_id} failed: {error}')

after_migrate runs only when something applied, so a no-op run stays quiet. migration_id is None when the run failed before reaching a migration. Only migrate calls before_migrate and after_migrate. on_error also runs when down fails. rehearse skips all three because a rehearsal leaves the real database unchanged.

Recover from a failed migration

A failed migration leaves partial changes behind only when nothing rolled them back. On an engine with transactional DDL, such as SQLite, Postgres, DuckDB, or SQL Server, a failed migration rolls back as a whole and Sustained writes no row, so you fix the migration and run migrate again. Sustained writes a failed-attempt row when the migration has transactional=False or the engine commits each DDL statement as it runs, as MySQL and Athena do. Validation then blocks the next run.

$ sustained validate
problem  migration '004_trim' has a failed attempt on record; clean up any
         partial changes, then run repair() and retry

Clean up whatever the failed statements left behind, then clear the bookkeeping:

$ sustained repair
repaired removed the failed attempt of '004_trim'

repair fixes only the tracking rows, so it does not undo half-applied schema changes or tell you which ones there were. Fix the schema by hand and check it before you run repair.

A failed down is recorded the same way when nothing rolled it back, on an engine without transactional DDL such as MySQL or in a migration with transactional=False. down then refuses to run until the row is cleared. Finish the revert by hand, then run repair, and the migration is pending again. If you instead restore the migration’s changes by hand, run repair and then baseline with that migration’s id, so it is recorded as applied.

Accept an edit to a migration that already ran

Checksums cover the exact SQL text, so reformatting counts as an edit and validation reports a mismatch. When the edit was deliberate:

$ sustained repair
repaired updated the stored checksum of 'create_venues'

Repeatables are the exception. For a repeatable, a changed checksum schedules a re-run, so repair leaves the checksum of a changed repeatable alone and the next migrate runs the new contents.

Deploy from two machines at once

You do not need to configure anything. During a run, the migrator keeps an exclusive advisory lock named after the tracking table, so a second deploy waits instead of racing. Postgres uses pg_advisory_lock, MSSQL uses sp_getapplock, and MySQL uses GET_LOCK. The last two return a status rather than raising, so the run stops with MigrationError when the status says the lock was not granted. SQLite and DuckDB serialize writers themselves.

Athena has no locking mechanism, so make sure that only one migrator runs at a time there.

Connections and drivers

Match the driver to the dialect

The connection’s parameter style must match the dialect’s placeholder, or execution fails at the driver.

Dialect Driver Placeholder
DEFAULT sqlite3 ?
POSTGRES psycopg, psycopg2 %s
MSSQL pyodbc ?
PRESTO trino ?
ATHENA pyathena with pyathena.paramstyle = "qmark" ?
MYSQL PyMySQL, mysqlclient %s
DUCKDB duckdb ? 
from sustained.dialects import Dialects

Model.set_dialect(Dialects.POSTGRES)
Model.bind(psycopg.connect('dbname=app'))

Set the dialect on Model to cover every model, or on a subclass to scope it.

Reuse connections across threads

from sustained.pool import ConnectionPool

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

Bind the pool the way you would a connection. Each statement checks one out for its duration; a transaction() block pins one for the whole block. An exhausted pool raises PoolTimeout after its timeout.

Run queries under asyncio

from sustained.aio import DbApiAsyncAdapter

adapter = DbApiAsyncAdapter(sqlite3.connect('tour.db', check_same_thread=False))
Model.bind_async(adapter)

shows = await Show.query().where('sold_out', '=', True).arun()

DbApiAsyncAdapter runs a synchronous driver in a worker thread. AiosqliteAdapter and AsyncpgAdapter use their native drivers instead. The async path matches the sync one: eager loading covers dotted paths and through relations, and nested transactions use savepoints.

Take the SQL and run it yourself

sql, params = Show.query().where('sold_out', '=', True).to_sql()
# ('SELECT * FROM shows WHERE sold_out = ?', (True,))

cursor.execute(sql, params)

Sustained never opens a connection on its own, so this path works with any pool, proxy, or framework session you already have.

Debugging

See the SQL a query will run

print(query)              # values inlined, for reading
print(query.to_sql())     # placeholders and parameters, as executed

str(query) output is for logs and eyeballs. Never send it to a database, because the inlined values are formatted for reading rather than for safe execution.

Log every statement the application runs

from sustained.execution import set_statement_listener

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

The listener fires after every executed statement, and it is global rather than per model or per connection. Pass None to remove it.

Read the query plan

Show.query().where('sold_out', '=', True).explain()
Show.query().where('sold_out', '=', True).explain(analyze=True)

analyze=True executes the statement to measure it, so on a write statement it changes data. MSSQL raises DialectError, so on MSSQL use SET SHOWPLAN_XML through raw SQL.

Catch a column typo before the database does

Declare the model’s columns with tableColumns, and the model gets this check automatically.

class Show(Model):
    tableName = 'shows'
    columns = ('id', 'venue_id', 'title', 'starts_at', 'sold_out')

Show.titel   # AttributeError, listing the declared columns

Without a declaration, every attribute resolves to a column name, and a typo reaches the database as a bad identifier.

Find out which migration failed

Sustained tags the driver’s exception with the id before re-raising it:

try:
    migrator.up()
except Exception as error:
    print(getattr(error, 'migration_id', None))

From the shell the same information is on stderr: error in '004_trim': column "legacy" does not exist.