Greyhorse SqlAlchemy library

Greyhorse framework library for SQLAlchemy support (sync and async, on PostgreSQL, MySQL/MariaDB and SQLite).

A database engine becomes a managed resource of the greyhorse framework — built, started, health-checked, repaired and torn down by it — plus transactional connection and session contexts, a repository over them, and migrations declared next to the engine as a transport that reads its DSN, with file-based profiles still supported as a second road.

Requires Python 3.14+ and SQLAlchemy 2.0.

Install

Pick the driver extras you need; the base package brings none of them.

pip install 'greyhorse-sqla[pg]'         # PostgreSQL: asyncpg + psycopg2
pip install 'greyhorse-sqla[sqlite]'     # SQLite: aiosqlite
pip install 'greyhorse-sqla[mysql]'      # MySQL/MariaDB: aiomysql + pymysql
pip install 'greyhorse-sqla[migration]'  # the `migration` CLI (alembic)

Usage

Every snippet below is a runnable program. The longer, commented versions live in examples/ and are executed by the test suite, so they cannot rot silently.

One engine, one connection

from sqlalchemy import text

from greyhorse.run import wrap_sync
from greyhorse.strand import running
from greyhorse_sqla import EngineConf, SqlaSyncConnCtx, SqlaSyncModule, SqlEngineType


def main() -> None:
    conf = EngineConf(type=SqlEngineType.SQLITE, dsn='sqlite:///:memory:')

    with running(SqlaSyncModule, args={EngineConf: conf}) as module:
        conn_ctx = module.get(SqlaSyncConnCtx).unwrap()
        with conn_ctx as conn:
            print(conn.execute(text('SELECT 1')).scalar_one())


wrap_sync(main)

The engine is created, its pool started, one query served, and everything stopped — on the way out of the with.

Async

Same shape, different border. The DSL declaration does not change: only the module, the context type and the await do. The plain sqlite:// DSN is enough — AsyncSqlaEngineFactory swaps in the aiosqlite driver itself, the same way postgresql:// becomes postgresql+asyncpg://.

from sqlalchemy import text

from greyhorse.run import run
from greyhorse.strand import running
from greyhorse_sqla import EngineConf, SqlaAsyncConnCtx, SqlaAsyncModule, SqlEngineType


async def main() -> None:
    conf = EngineConf(type=SqlEngineType.SQLITE, dsn='sqlite:///:memory:')

    with running(SqlaAsyncModule, args={EngineConf: conf}) as module:
        conn_ctx = module.get(SqlaAsyncConnCtx).unwrap()
        async with conn_ctx as conn:
            result = await conn.execute(text('SELECT 1'))
            print(result.scalar_one())


run(main)

The pieces, and why you will usually want them directly

SqlaSyncModule/SqlaAsyncModule above are convenience wrappers for the single-storage case. The library's real API is the three pieces they bundle:

An application that needs several storages lists the pieces it wants from each library on its own module — no subclassing, no multiple inheritance. See examples/03_multi_storage.py.

Connections and sessions differ on re-entry

Both products are outcome contexts: apply() commits, and a forgotten apply() or an exception rolls back. They part ways when a borrow is RE-ENTERED — a helper or repository opening the same context inside an outer borrow:

A SQLAlchemy Session.commit() settles everything the session has done — it has no notion of "just my scope" — so a nested apply() there would publish the outer borrow's work and leave it committed if the outer operation later failed. Refusing is loud and safe: it changes nothing and does not consume the borrow, so the outer scope can still apply() or cancel(). If you need nested units of work with independent outcomes, use connection(), or take a separate session.

When cleanup itself fails

A borrow ends by doing things you never asked for by name: rolling back a transaction that was not applied, closing the transaction object, returning the connection to the pool, closing the session. The rule for when any of that fails:

So the exception that reaches you is the one that caused the unwind — your own domain error — never a secondary failure of the package's tidying up. You do not need a try/except around a borrow to protect the error you already have.

The warning names the engine, the operation (rollback, transaction-close, connection-release, session-close) and the exception's TYPE. It deliberately carries neither the exception's text nor a traceback: a driver's connection error routinely quotes the DSN back, password included. The DSN in the line is redacted (greyhorse.data.redact_dsn). Full reasoning in greyhorse_sqla/cleanup.py.

Health and repair

The border's check() asks a real connectivity question (SyncSqlaEngine.is_alive() / AsyncSqlaEngine.is_alive()), not a start/stop counter — so a database that has gone away is reported as such and the framework can repair the resource. Probes are single-flight per DSN and briefly cached, so a health check on every tick does not turn into a connection storm while the database is down.

Migrations

migration --help

Alembic under the hood — importable on a base install with neither typer nor alembic present; the migration extra brings both. Two equally valid roads to a runnable migration, chosen by which flag you pass:

As a transport (--app)

Declare a migration set next to the engine it migrates, in its own component — it reads the DSN from that engine's EngineConf, so nothing duplicates the password into a second file:

from pathlib import Path
from typing import ClassVar

from greyhorse.strand import Component, Shared
from greyhorse_sqla import Migrations, SyncSqlaEngine
from greyhorse_sqla.migration.handlers import SqlaMigrations


class OrdersMigrations(SqlaMigrations):
    alembic_path: ClassVar = Path(__file__).parent / 'alembic'
    metadata_package: ClassVar = 'app.orders.models'


class OrdersMigrationsComponent(Component):
    imports: ClassVar = Shared[SyncSqlaEngine]
    exports: ClassVar = OrdersMigrations
    handlers: ClassVar = Migrations(OrdersMigrations, name='orders')

SqlaMigrations lives at that longer greyhorse_sqla.migration.handlers path, not at the top of the package, because it reaches alembic — absent from a base install — and importing it must stay opt-in.

Two rules this shape imposes, both consequences of every module tree in an Application needing a Gateway for each transport it carries:

1. Migrations live in their OWN component, apart from anything serving HTTP — one component carrying both drags an HTTP gateway (and its port) into every target that mounts it, migration up included.
2. Share the storage module between your production target and a migration target with a SubModule row, never by declaring the set twice.

A target — the place a real Application gets built, one per run mode — owns the gateway and the configuration; the CLI never reads a DSN on this road, and never fetches or substitutes args for you:

# app/targets/migrations.py
def build() -> Application:
    return Application(MigrateTarget, gateways=(MigrationGateway(),), args={...})

migration up --app app.targets.migrations:build
migration up --app app.targets.migrations:build --only orders

ATTR is either a ready Application or a zero-argument callable returning one — prefer the callable form, since importing the module must not itself open a pool (--help has to work without a database). examples/05_migrations.py runs both rules end to end, upgrading through one target and dispatching HTTP through another that shares the same storage submodule.

File profiles

The other road, unchanged: a single profile (--dsn/--alembic-path/ --metadata) or a TOML set of profiles (--config, plus --only NAME[,NAME] to filter) drives MigrationRunner directly — no Application/Module in sight. Reach for this when the caller has nothing but a DSN, e.g. a deploy image with no application tree to import.

Autogenerate is scoped either way: the generated env.py only ever considers tables that belong to your own metadata, so it will not propose dropping another application's tables sharing the database.

Schemas are created for you on PostgreSQL — every schema your metadata declares, whether on the MetaData itself or per-table through __table_args__ = {'schema': ...}. Where alembic's own alembic_version table lands follows metadata.schema; when the metadata is schema-less but its tables carry schemas, nothing can be inferred, so name one explicitly:

[[profiles]]
name = "reporting"
dsn = "postgresql://user:pass@host/db"
alembic_path = "alembic/reporting"
metadata = "app.models.reporting:metadata"
version_table_schema = "reporting"   # or --version-table-schema

Without it two schema-scoped applications in one database share a single public.alembic_version, and since revisions are numbered by file count every project's first revision is 001 — the second application either skips its own initial migration or cannot find 001 in its own scripts.

Examples

examples/ holds runnable programs, ordered so each adds one idea:

uv run python examples/01_minimal.py

They are covered by tests/test_examples.py, which runs each as a subprocess and asserts the exact lines it prints — so they cannot rot into prose that lies.

Development

# The extras are not optional for development: a bare `uv sync` REMOVES
# them, and the migration tests stop collecting without `typer`. `pg` is
# left out on purpose -- it pulls `psycopg2` from source, which needs a
# local libpq; the dev group brings `psycopg2-binary` instead.
uv sync --extra sqlite --extra mysql --extra migration

uv run pytest                      # sqlite + unit tests, no servers needed
uv run ruff check && uv run ruff format
uv run mypy greyhorse_sqla/ examples/

Postgres and MySQL tests are gated behind SQLA_TEST_POSTGRES_URI and SQLA_TEST_MYSQL_URI; unset, they skip. tests/docker-compose.yml brings up both.

License

MIT.