Sustained.py

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

Your models declare their columns. Sustained diffs them against the live database, writes the migration, and runs the whole thing forwards and back on a rehearsal before it touches the real schema:

$ sustained rehearse
rehearsed 003_sessions  up ok, down ok
rehearsed 004_trim      up ok, down ok
rollback complete, database unchanged

The same model classes build and run your queries:

adults = User.query().where(User.c.age >= 18).orderBy('name').run()

Migrations

• Generated from your models. Migrator.sync(models) diffs the live database, generates the migration, records it, applies it, and down() rolls it back. Only the difference is applied on each run.
• Rehearsed before they land. sustained rehearse applies every pending migration, runs the down steps back down, and rolls the whole thing back. A migration that does not run, or does not reverse, says so while the real schema is still untouched. Point it at a scratch database when the rollback cannot be trusted.
• Planned in one screen. sustained plan merges pending migrations, validation problems, and model drift, labels destructive statements, and exits 2 when work is waiting. --json for a pipeline.
• Checked, not trusted. The tracking table holds a sequence number, a SHA-256 checksum, timing, and a success flag per migration. validate() blocks a run when a migration was edited after it ran, arrives out of order, or left a failed attempt; repair() fixes the bookkeeping.
• Safe by refusal. Drops need allow_drops=True, renames need hints, NOT NULL changes need a backfill. Constraint drift is reported, never silently migrated.
• Yours to write. Migrations can be Python objects, <id>.up.sql and <id>.down.sql files with ${placeholders}, or <id>.repeat.sql files that re-run whenever their contents change. script('up') renders every statement for offline review.
• Ready for deploys. The sustained console script runs plan, status, rehearse, migrate, down, validate, repair, script, and baseline from the shell, with exit codes and config module callbacks around each run. Concurrent deploys queue on an advisory lock. baseline adopts a database that already has the schema.

What else it does

• SQL building for the default (ANSI), Postgres, MSSQL, Presto, AWS Athena, and DuckDB dialects: joins, CTEs (including recursive), unions, window functions, CASE expressions, and subqueries. Features a dialect lacks raise DialectError at build time.
• Safe execution: every statement runs parameterized. Transactions nest through savepoints. update() and delete() refuse to run without a WHERE clause.
• Writes: insert(), update(), delete(), upserts with onConflict(), INSERT ... SELECT, CTAS, and RETURNING.
• Typed filters: User.query().where((User.c.age > 21) & User.c.name.like('A%')).
• Results as model instances, dicts, pandas DataFrames, or pyarrow Tables, with withGraphFetched() eager loading.
• Async: the same queries run through driver adapters (asyncpg, aiosqlite, or any sync driver in a worker thread) with await query.arun(), including an AsyncMigrator.

What it does not do

No lazy loading, no dirty tracking or save(), no identity map, no result caching, no cross-dialect emulation of missing features, and no guessed migrations: drops, renames, and NOT NULL backfills all require explicit opt-ins or hints. Writes and schema changes only happen when you spell them out.

Installation

python3 -m pip install sustained

Usage

from sustained import Model, RelationType

class Person(Model):
    tableName = 'persons'

class Animal(Model):
    tableName = 'animals'
    relationMappings = {
        'owner': {
            'relation': RelationType.BelongsToOneRelation,
            'modelClass': Person,
            'join': {
                'from': 'animals.ownerId',
                'to': 'persons.id'
            }
        }
    }

# Build a query
query = Animal.query().select('animals.name', 'persons.name').leftOuterJoinRelated('owner')

print(query)
# SELECT animals.name, persons.name
# FROM animals
# LEFT OUTER JOIN persons
#   ON animals.ownerId = persons.id


# Execute against any DB-API 2.0 connection
import sqlite3

conn = sqlite3.connect('app.db')
Animal.bind(conn)

# Parameterized execution with model hydration
animals = Animal.query().where('species', '=', 'dog').orderBy('name').run()

# Or take the SQL and parameters and execute them yourself
sql, params = Animal.query().where('species', '=', 'dog').to_sql()
# sql:    "SELECT * FROM animals WHERE species = ?"
# params: ('dog',)

Models carry their own schema, so a column change is a migration:

from sustained.migrations import Migrator
from sustained.schema import Integer, String, Text

class User(Model):
    tableName = 'users'
    tableColumns = {
        'id': Integer(primary_key=True, autoincrement=True),
        'email': String(120, unique=True, nullable=False),
    }

migrator = Migrator(conn, [])
migrator.sync([User])              # creates the users table

User.tableColumns['bio'] = Text()
migrator.plan([User])              # the migration sync() would generate
migrator.sync([User])              # adds only the bio column
migrator.down()                    # rolls it back

From the shell, a config module names the connection, the migrations directory, and the models:

# sustained_config.py
import sqlite3

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

migrations_dir = 'migrations'
models = [User]

$ sustained plan        # pending migrations, validation problems, model drift
$ sustained rehearse    # run it all, forwards and back, then roll back
$ sustained migrate     # apply it for real
$ sustained down        # --steps N or --to ID

See Schema and Migrations for SQL file migrations, repeatables, checksum validation, baseline, and the Athena rules.

Documentation

The documentation covers schema and migrations at length, plus models, queries, dialects and drivers, filtering, grouping, relations and joins, execution, pooling, and async, and schema and migrations. The API reference lists every public method by task.

Development

To install from source:

git clone https://github.com/wetherc/sustained.git
cd sustained
python3 -m pip install -e .

This project uses pre-commit to format code, lint, type check, and run the test suite before each commit:

pip install pre-commit
pre-commit install