---

Why?

An application with a database needs the same things every time: an engine per database, a session per request or message, a way to change the schema one reviewed step at a time, and commands to run those steps from a deploy script. Each is a few dozen lines of glue, written again in every project.

This package is that glue, done once. You describe each connection — a URL, its replicas, its engine and session options, where its revisions live — and get:

• 🔌 Engines and sessions built from one configuration, on SQLAlchemy's async API and advanced-alchemy's models, repositories and services.
• 📖 Read replicas — reads go to a replica, writes and everything after them to the primary.
• 🧬 Versioned migrations on Alembic: diffed from your models, run up or down to any revision, one at a time or all at once — with no env.py and no alembic.ini to maintain.
• 📜 A full history: every revision applied, when, and how long it took.
• 🧾 One transaction per message — with xtr-messenger, every handler of a message shares one session, committed once they all succeeded and rolled back otherwise.
• 🛠️ Fifteen console commands, from orm:database:create to orm:migrations:up-to-date.

Install

uv add "xtr-orm[postgres]"                        # asyncpg
uv add "xtr-orm[di,console,messenger,sqlite]"     # everything, on SQLite

Extra	Brings	For
postgres / mysql / sqlite	asyncpg / asyncmy / aiosqlite	The async driver of your database
di	xtr-dependency-injection	The OrmBundle: engines, sessions, migrators from one configuration
console	xtr-console	The orm:* commands
messenger	xtr-messenger	The message bus middleware

Quick start

Without a container, a Migrator runs one database's revisions:

from sqlalchemy.ext.asyncio import create_async_engine
from xtr_orm import Migrator, MigrationsConfig

from app.models import Base

engine = create_async_engine("postgresql+asyncpg://app:secret@db/shop")
migrator = Migrator(engine, MigrationsConfig(directory="migrations"), Base.metadata)

await migrator.diff("add the catalogue")  # writes migrations/<date>_<rev>_add_the_catalogue.py
await migrator.migrate()  # runs every revision not applied yet
status = await migrator.status()

Migrations

Revisions are ordinary migration files, one per change, in the connection's migrations directory. The package writes them from its own template and runs them itself, handing the migration runner the live connection and your table definitions directly — there is no start-up script to keep in step with your models.

The database keeps two tables, in the same schema:

• the version table (alembic_version) holds the revisions the database is at. It decides what a migration runs.
• the history table (alembic_version_history) holds a row per revision ever applied — version, executed_at, execution_time in milliseconds. Each row is written by the step that applies its revision, in the same transaction, so the two agree. It is what lets a report name every applied revision, including one whose file is gone.

A target — for migrate and plan — is a revision (a full or partial id), a relative one (+1, -2, ae10+1), or latest, first (before every revision), next, prev or current. A revision already applied is migrated down to; one that is not, up to.

Migrator method	Does
status()	Where the database stands: current, latest, previous, next, available, executed, new, and executed but without a file
plan(target) / migrate(target) / migrate_sql(target)	What reaching a target runs / run it / the SQL it would run, recording included
plan_for_versions(versions, direction) / execute(...) / execute_sql(...)	What running exactly those revisions, each up or down, runs — every id resolved and checked / run them, nothing before or after / the SQL it would run
generate(message)	Write an empty revision after the latest
diff(message, allow_empty=, from_empty_schema=)	Write what differs between the database and the table definitions
dump_schema(message, table_filters=)	Write a first revision creating every table an existing database has
add_versions(versions) / add_all_versions()	Record revisions as applied, running nothing
delete_versions(versions) / delete_all_versions()	Record revisions as not applied, running nothing
rollup()	Forget every applied revision, then record the single remaining one

Running one revision alone. The version table holds only where the database is, so a revision can be run — or recorded — on its own only where that stays true: going up, it must not be applied and must follow only applied revisions; going down, it must be applied and followed by none that is. Several versions are checked in turn, so a revision and the one after it can go together. Anything else is refused before the database is touched.

Up to date means every revision with a file is applied — the database's versions against the files, not its schema against your models; diff compares those.

MigrationsConfig:

Field	Default	Meaning
directory	"migrations"	Where revision files are read and written; a relative path is relative to the working directory — under a kernel, start it with %kernel.project_dir%
version_table / history_table	"alembic_version" / "alembic_version_history"	The two tables
version_table_schema	None	Their schema; the connection's default when None
file_template	"{year}{month}{day}{hour}{minute}{second}_{rev}_{slug}"	How a file is named; also {epoch}
truncate_slug_length / timezone	40 / None	The longest message part of a name / the zone of a revision's date
template_directory	None	A directory with a script.py.mako to render new revisions from
recursive_version_locations	False	Read subdirectories of directory too
post_write_hooks	()	Programs run on each file written, such as a formatter
transaction_per_migration / transactional_ddl	False / None	Commit after each revision / run schema changes in the transaction
compare_type / compare_server_default	False / False	What a diff compares besides tables, columns, indexes and constraints
render_as_batch	False	Write table changes as batches, which SQLite needs
include_schemas / include_name / include_object	False / None / None	What a diff reads
render_item / process_revision_directives / user_module_prefix	None	How a diff is written
context_options	{}	Any other migration context option, passed as it is

A custom column type is written under its own module's name, and the revision imports that module — advanced_alchemy.types.guid.GUID(length=16) with import advanced_alchemy.types.guid; set user_module_prefix or render_item to write it otherwise. compare_type is off by default because a type the database stores as another — a UUID kept as bytes on SQLite — would otherwise differ in every diff.

Your own revision template

New revisions are rendered from script.py.mako — the package's own, or the one in template_directory. It receives what Alembic's own templates do: message, up_revision, down_revision, branch_labels, depends_on, create_date, imports, upgrades, downgrades, and the comma filter (${down_revision | comma,n}). A template written for an Alembic project works as it is; nothing but that file is read from the directory.

MigrationsConfig(
    template_directory="%kernel.project_dir%/resources/migrations",
    user_module_prefix="sa.",  # a template aliasing sa.GUID = GUID, as advanced-alchemy's does
)

A template calling op.get_context().autocommit_block() commits a revision's schema changes before the version and history tables are written, and gives up the transaction a failing revision would roll back; set transaction_per_migration=True with it, and keep it for the revisions that need it (CREATE INDEX CONCURRENTLY).

Use in an application

Everything adding this package to an application on xtr-dependency-injection takes — and, read backwards, what removing it undoes.

• Install — uv add "xtr-orm[di,console,postgres]"; mysql or sqlite instead of postgres for those databases; drop console to go without the commands; add messenger for the message bus middleware.
• Recipe — uv run xtr-recipes recipes:sync does the Activate, Configure and Environment steps below: it lists OrmBundle, writes a starting <app>/config/orm.py with the default connection, and DATABASE_URL in .env. It prints the middleware step, which a recipe cannot make for you.
• Activate — OrmBundle: {"all": True} in BUNDLES in <app>/bundles.py, imported from xtr_orm.bundle.
• Brings along — the logging and console bundles, when those packages are installed. With the messenger bundle active too, the middleware are registered for a bus to list.
• Configure — the recipe writes a starting <app>/config/orm.py with the default connection reading DATABASE_URL, its revisions in migrations in the project directory and diffs against the models advanced-alchemy registers. Add connections, replicas and table definitions to that @configure function returning OrmConfig — see Kernel / bundle.
• Environment — DATABASE_URL, read when a connection is first used — engine, session, migrator or command — not when the application boots. Its query may carry options; see Options in the URL.
• Ignore — nothing: commit migrations/.
• Remove — take the orm_* entries out of every bus's middleware, drop the BUNDLES entry, delete <app>/config/orm.py, then uv remove xtr-orm.
• Check — debug:bundles shows orm as listed and active, and orm:migrations:status reaches the database.

Kernel / bundle

# app/bundles.py
from xtr_orm.bundle import OrmBundle

BUNDLES = {OrmBundle: {"all": True}}

# app/config/orm.py
from xtr_dependency_injection import configure, env
from xtr_orm.bundle import ConnectionConfig, OrmConfig

from app.models import Base


@configure
def orm() -> OrmConfig:
    return OrmConfig(
        connections={
            "default": ConnectionConfig(
                url=env("DATABASE_URL"),
                replicas={"replica1": env("DATABASE_REPLICA_URL")},
                engine_options={"pool_size": 10, "pool_pre_ping": True},
                session_options={"expire_on_commit": False},
                metadata=Base.metadata,
            ),
            "reports": ConnectionConfig(url=env("REPORTS_URL"), bind_key="reports"),
        },
    )

For every connection, qualified by its name — and the default connection's without a qualifier too:

Service	What it is
SQLAlchemyAsyncConfig	advanced-alchemy's configuration, for its repositories and services
AsyncEngine	The engine on the primary database
async_sessionmaker[AsyncSession]	The session maker — RoutingAsyncSessionMaker instead for a connection with replicas
AsyncSession	One per unit of work — a request, a message, a command — closed, and rolled back if not committed, when the unit ends
Migrator / DatabaseManager	The connection's migrations / its database, both on the primary
ConnectionRegistry	Every connection by name, unqualified — what the commands use

from typing import Annotated

from sqlalchemy.ext.asyncio import AsyncSession
from xtr_dependency_injection import Injected, Target, as_service


@as_service
class InvoiceRepository:
    def __init__(self, session: Injected[AsyncSession]) -> None: ...


async def report(session: Annotated[AsyncSession, Target("reports")]) -> None: ...

Every engine is disposed when the container closes. With the logging bundle active, an orm channel is added and every migrator logs each revision it runs there. With the messenger bundle active, the message bus middleware are registered by name, and every message is a unit of work: its handlers share one AsyncSession per connection.

OrmConfig field	Default	Meaning
default_connection	"default"	The connection provided without a qualifier, and used by every command not given --connection
connections	{"default": ConnectionConfig()}	Every connection, by name

ConnectionConfig field	Default	Meaning
url	"%env(DATABASE_URL)%"	The database URL, with an async driver; its query may carry options — see below
replicas	{}	Read replicas, each a URL by name
keep_replica	False	Go back to the replicas once a session commits
engine_options	{}	echo, pool_size, max_overflow, pool_pre_ping, pool_recycle, isolation_level, connect_args, …
session_options	{}	expire_on_commit, autoflush, info, …
alchemy_options	{}	Any other option of advanced-alchemy's configuration, such as enable_touch_updated_timestamp_listener
bind_key	None	Which models belong to this database, when they declare several
metadata	None	The table definitions a diff compares with; advanced-alchemy's for bind_key when None
migrations	None	MigrationsConfig; migrations in the project directory for the default connection, migrations/<name> for any other, when None

Options in the URL

One environment variable can configure a whole connection. An engine or session option, or keep_replica, is taken out of the URL's query and merged over the one configured — the URL wins; every other option stays on the URL for the driver:

DATABASE_URL="postgresql+asyncpg://app:secret@db/shop?pool_size=20&expire_on_commit=false&ssl=require"

Read from the URL	Options
Engine	echo, echo_pool, enable_from_linting, hide_parameters, insertmanyvalues_page_size, isolation_level, label_length, logging_name, max_identifier_length, max_overflow, paramstyle, pool_logging_name, pool_pre_ping, pool_recycle, pool_size, pool_timeout, pool_use_lifo, query_cache_size, use_insertmanyvalues
Session	autobegin, autoflush, expire_on_commit, join_transaction_mode, twophase
Connection	keep_replica

Flags read 1/true/yes/on and 0/false/no/off; numbers are numbers; echo also takes debug. A value that does not read fails the service that needed the URL, naming the option. Only the connection's own URL is read this way, not its replicas'.

Read replicas

With replicas, a session reads from a replica picked at random, and goes to the primary — url — for a write, for a transaction's statements, and for every read after them, so it sees what it wrote. It stays there for the rest of its unit of work, unless keep_replica sends it back to the replicas once it commits. Each unit of work starts on the replicas, whatever the one before wrote. Migrations and the database commands always use the primary.

A block can be sent elsewhere with advanced-alchemy's primary_context() and replica_context(), from advanced_alchemy.routing.

Mapping tables from a bundle

A library can bring its own tables to an application's database — tables whose prefix, names or column types the application configures, and which therefore exist only once that configuration is read. OrmMappingsPass puts them into the table definitions of the connection the application picked, while the container is compiled: early enough for orm:migrations:diff to write them like any of the application's own models.

The library's bundle keeps the pass, registers it in build, and hands over the driver it builds from its configuration in load_extension:

from xtr_orm.bundle import OrmBundle, OrmMappingsPass


@final
@required_bundle(OrmBundle)
@as_bundle("invoicing", config=InvoicingConfig)
class InvoicingBundle(Bundle[InvoicingConfig]):
    def __init__(self) -> None:
        self._mappings = OrmMappingsPass("invoicing.connection", "invoicing.persist")

    @override
    def build(self, builder: ContainerBuilder) -> None:
        builder.add_compiler_pass(self._mappings)

    @override
    def load_extension(
        self, config: InvoicingConfig, services: ServiceConfigurator, builder: ContainerBuilder
    ) -> None:
        builder.set_parameter("invoicing.connection", config.connection)
        builder.set_parameter("invoicing.persist", config.persist)
        if config.persist:
            self._mappings.set_driver(InvoicingDriver(config.table_prefix))

A driver is anything satisfying MappingDriverInterface: one call, adding its tables and naming them. Subclass it to have the type checker hold you to that call.

from xtr_orm import MappingDriverInterface


@final
class InvoicingDriver(MappingDriverInterface):
    def __init__(self, prefix: str) -> None:
        self._prefix = prefix

    def load_metadata(self, metadata: MetaData) -> tuple[str, ...]:
        name = f"{self._prefix}invoice"
        if name in metadata.tables:
            return ()
        _ = Table(name, metadata, Column("id", Integer, primary_key=True))
        return (name,)

• Which definitions — the ones the connection's migrations compare the database with: its metadata when it has one, else the ones advanced-alchemy registers for its bind_key. A connection keeping several sets is refused, naming it; give that one a single MetaData.
• Idempotent, always — load_metadata skips the tables already there and leaves them out of what it returns. The registered definitions are process-wide — advanced-alchemy keeps one MetaData per bind key for the whole process — so two kernels in one process load the same driver into the same one. That outlives the flag: a kernel built with the library's persistence off maps nothing itself, yet still sees every table a persisting kernel of the same process put in that bind key, or in a module-level MetaData both reach, and diffs them as its own. Give such a kernel a bind_key, or a freshly built MetaData, that no persisting kernel used.
• On or off — the flag parameter is read for truth, and an unset one means off, so a library whose persistence an application turned off maps nothing. On with no driver handed over is refused, as is a connection name nothing configured — both naming what is missing.
• Read while compiling — both parameters are read before %name% references are resolved and long before an env(...) one is, so a bundle sets them from its own resolved configuration: a plain name and a plain flag. One still holding a %name% reference is refused, naming the parameter, as is a connection parameter holding something that is not a name at all — a name is never coerced out of it, since a connection may be called 7.

Console commands

Every command takes --connection NAME; the default connection's is used without it. The commands that change the database ask first; -n answers yes.

Command	Does
orm:database:create [--if-not-exists]	Creates the database, through the server's maintenance database
orm:database:drop --force [--if-exists]	Drops it; refused without --force
orm:run-sql SQL [--force-fetch]	Runs a statement, committed; shows its rows or how many it changed — when the driver can tell
orm:migrations:current	The revisions the database is at — base for none
orm:migrations:latest	The revisions no other follows
orm:migrations:list	Every revision with its status, when it ran and how long it took
orm:migrations:status	The tables, the database, previous / current / next / latest, and the counts
orm:migrations:up-to-date [-u] [-l]	Exits 1 with revisions not applied; 2 with --fail-on-unregistered and applied revisions without a file
orm:migrations:migrate [TARGET] [--dry-run] [--write-sql PATH] [--allow-no-migration]	Runs every revision to TARGET, latest by default
orm:migrations:execute VERSION... [--up | --down] [--dry-run] [--write-sql PATH]	Runs exactly those revisions
orm:migrations:generate [--message M]	Writes an empty revision
orm:migrations:diff [--message M] [--allow-empty-diff] [--from-empty-schema]	Writes what the models add; the database must be up to date
orm:migrations:dump-schema [--message M] [--filter-tables RE]...	Writes a first revision from an existing database; rollup then records it
orm:migrations:version [VERSION] (--add | --delete) [--all]	Records revisions as applied or not, running nothing
orm:migrations:rollup	Records the single remaining revision as all there is

--write-sql writes to the file named, or to a dated file in the directory named.

Every command takes the ConnectionRegistry it acts on as its one constructor argument — the bundle's with a container, and your own without one. Nothing stands in for it: a console with no container to supply it reports the parameter it cannot fill.

from xtr_orm import ConnectionRegistry, DatabaseManager
from xtr_orm.command import MigrationsMigrateCommand

registry = ConnectionRegistry()
registry.register("default", engine=engine, migrator=migrator, database=DatabaseManager(url))

migrate = MigrationsMigrateCommand(registry)

Message bus middleware

With the messenger extra and a kernel running both bundles, a message handler asks for a session the way any service does, and three middleware — registered by name — decide what happens to it. Nothing is added to a bus by itself: the bus lists the ones it wants.

# app/billing/handlers.py
@as_message_handler(IssueInvoice)
async def issue(message: IssueInvoice, session: Injected[AsyncSession]) -> None:
    session.add(Invoice(order_id=message.order_id))  # no commit: orm_transaction does it


@as_message_handler(IssueInvoice)
class NotifyAccounting:
    # A scoped repository on the same session: one transaction for both handlers.
    async def __call__(
        self, message: IssueInvoice, invoices: Injected[InvoiceRepository]
    ) -> None: ...

# app/config/messenger.py
MessageBusConfig(
    middleware=[
        "orm_close_connection",
        "orm_transaction",
        {"orm_transaction": {"connection_name": "reports"}},  # another connection too
        "orm_open_transaction_logger",
    ],
)

Name	Class	Does	Arguments
orm_transaction	TransactionMiddleware	Runs every handler of a message in one transaction, committed once they all succeeded and rolled back otherwise; a failure drops the other handlers' handled stamps, so a retry runs them all again	connection_name, the default connection when left out
orm_close_connection	CloseConnectionMiddleware	Closes the connections once a worker handled a message it consumed, so an idle worker holds none — once the last, when it handles several at once	connection_names, every connection in use — its engine built — when left out
orm_open_transaction_logger	OpenTransactionLoggerMiddleware	Logs an error — to the application's logger when the logging bundle is active — when a handler began a transaction — begin(), begin_nested() — and left it open, in a message that was handled	connection_names, every connection in use — its engine built — when left out

• One session per message. The messenger bundle makes every message a unit of work, and a scoped service — the session, a repository built on it — is the unit's: every handler of the message gets the same one, released once the message is done with. A message dispatched while another is handled joins its unit, and its transaction — in a savepoint of its own, so when it fails only its handlers' work is rolled back, and the handler that dispatched it decides whether the rest goes on.

• Handlers do not commit under orm_transaction; a handler that does ends the transaction early, and one using async with session.begin() is refused, the transaction being open already.

• Order matters. orm_close_connection outermost, so it closes after the transaction ended; orm_open_transaction_logger inside orm_transaction, so it looks before the commit closes everything.

• Only messages a worker consumed are closed after — one carrying a ReceivedStamp on arrival. A message handled during its dispatch, sync:// included, closes nothing. A worker handling several at once closes once the last of them is done, keeping its pool meanwhile.

• Follow-ups wait for the commit when stamped. A message a handler dispatches to a transport is sent at once — before the commit, and whether or not it comes. Dispatch it with DispatchAfterCurrentBusStamp and it goes out only once the message being handled was committed, and never if it was rolled back:

  await bus.dispatch(SendReceipt(order.id), DispatchAfterCurrentBusStamp())
  

• A dropped connection is found by the engine option pool_pre_ping as a connection is taken from the pool — ?pool_pre_ping=true on a worker's URL — so there is no middleware for that.

Without a container, build them with a ConnectionRegistry whose connections are registered with a session — what returns the session the handlers use — and put the instances in the configuration.

Creating and dropping databases

DatabaseManager(url) creates and drops the database a URL names. A database cannot be created from a connection to itself, so the server is reached through its maintenance database — postgres on PostgreSQL, master on SQL Server, none on MySQL and MariaDB — with every statement committed as it runs, and with every connection parameter of the URL but its database: the credentials, the host and the driver's query options, such as ssl. A SQLite database is its file. Any other server is refused with UnsupportedDatabaseError.

create and drop let the server refuse a database that is already there or is not, which is why both take a flag — create(if_not_exists=True), drop(if_exists=True) — to be told so instead. On SQLite the file is the database: dropping one that is not there raises MissingDatabaseError.

Errors

Every error derives from OrmError.

Error	Raised when
InvalidArgumentError	A configuration cannot be used — an empty name, an unknown option, a URL option that does not read (also a ValueError)
MigrationError	A migration is refused or its revisions cannot be read — an unknown or ambiguous version, a revision out of order, a diff of a database behind its revisions
MissingDatabaseError	A database is dropped that is not there — a SQLite file that does not exist — without if_exists (also a FileNotFoundError)
UnknownConnectionError	A connection is asked for by a name nothing registered (also a LookupError)
UnsupportedDatabaseError	A database is created or dropped on a server whose statements are not known
SessionUnavailableError	A connection's session is asked for outside a unit of work, or of a connection registered without one

Known limitations

• Migrations take no lock. Two processes migrating one database at once — two deploys starting together — both run the same revisions; run orm:migrations:migrate from one place, before the application starts.
• Applied times follow the database. executed_at has a time zone on PostgreSQL and is naive elsewhere: UTC on SQLite, the session's time zone on MySQL.
• advanced-alchemy's model registry is process-wide. Its metadata_registry maps a bind_key to table definitions for the whole process, so two kernels in one process share it. Give a connection its metadata explicitly to keep a kernel's diff to its own models.
• The version table holds only the current revisions, so a revision is run or recorded alone only where the database stays consistent — see Migrations. The history table records every one.
• Async only: engines and sessions are async; there is no sync counterpart.

Development

Developed in the python-xtr monorepo, under packages/xtr-orm; run the commands below from there. Every test runs on SQLite files in the test's own directory; no server is needed.

uv sync
uv run ruff check && uv run ruff format --check && uv run basedpyright && uv run ty check && uv run pytest

License

MIT — see LICENSE.