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.

A repository behind its own fragment

The canonical domain layering — a repository that takes a session context and never commits, an activity that holds the repository as a plain field, an API that holds the activity the same way — can live entirely on ONE Fragment, with only the top-level API named in exports. The component's own providers = SqlaSyncSessions still reaches the repository's session hole, even though that hole now sits two layers below the exported product instead of on the export's own constructor. See examples/06_fragment_repository.py.

Fragments as layers, components as slices

The SAME domain layering can also be split across SEVERAL fragments — one per layer — and then reused, unchanged, by several components, each taking its own slice through the stack (e.g. a "widgets" component and a "gadgets" component sharing one repository fragment, one activity fragment, and one API fragment). Every hop between layers then crosses a fragment boundary inside one component. providers = SqlaSyncSessions still reaches the repository layer's session hole, now two fragment boundaries below the exported product. See examples/07_matrix_fragments.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.

Sharing one session across repositories (joined_session()/joined_connection())

Two repositories, each holding its own SqlaAsyncSessions.session(), do NOT share a transaction just because they run inside the same request_window(). session() builds a brand new context object on every call, and a joined context's default key is that object's own identity — so each repository ends up with its own independent RequestWindow root, its own independent commit/rollback decision, even though both are touching the same engine on the same request.

joined_session()/joined_connection() close that gap by keying the join to the ENGINE instead of to whichever context object happens to be asking. A second repository's joined_session() call over the same engine lands on the root the first one already opened, so every repository on the SESSION road makes one acquisition and one commit/rollback decision, no matter how many of them touch that engine — and the same holds, separately, for every repository on the CONNECTION road calling joined_connection():

notes_connection = sessions.joined_connection   # instead of sessions.connection

The two roads do not join each other. joined_session() and joined_connection() are keyed apart on purpose (the provider's own _joined_key() folds in 'session'/'connection' as well as the engine), so one request that mixes both — one repository on joined_session(), another on joined_connection(), same engine — still gets TWO independent roots, each with its own apply()/cancel() decision. Put every repository over one engine on the SAME road if you need them to share one outcome.

Who switches this — the composition root, not the repository. A repository is written against the ordinary SqlaAsyncSessionCtx/ SqlaAsyncConnCtx type and has no idea joined_*() exists. JoinedSqlaAsyncSessionCtx/JoinedSqlaAsyncConnCtx are real subtypes of those ordinary types, so the substitution is invisible to it; only the fragment/factory that BUILDS the repository switches from sessions.session to sessions.joined_session. Give the repository the SqlaAsyncSessions provider itself and let it call joined_*() on its own instead, and you lose the ability to construct that repository in a unit test against a bare context — its constructor now demands a provider, which demands a slot, which demands a real engine.

The practical form is a factory per parameter, not an open context: Callable[[], SqlaAsyncConnCtx], called fresh in every method, never opened once and reused. A context already open when a DIFFERENT asyncio.Task tries to enter it raises CompetingBorrowError rather than queuing — so a repository serving concurrent requests needs a fresh wrapper object per call, not one instance shared across them; the sharing that matters happens one layer down, keyed to the request window, not by reusing a Python object. NoteRepository in tests/e2e/test_web_gateway_over_real_postgres.py shows the full shape, including why the "joins the window" connection and the "runs outside any window" schema connection are kept as two separate factories rather than one.

Outside a request_window(), joined_*() is fully transparent: it hands back a plain, independent session()/connection(), and nothing about calling it instead of the plain method changes.

What looks equivalent and is not: into_joined(sessions.session()) with no explicit key. The default key is the wrapped context's own object identity, and session() builds a new object on every call — so two such calls, even against the same engine, land on two independent roots. joined_*() exists precisely so callers never have to get this key right by hand.

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, not a start/stop counter — so a database that has gone away is reported as such and the framework can repair the resource.

The sync road probes directly: SyncSqlaEngine.is_alive() opens a connection and runs SELECT 1. 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.

The async road answers the same question passively, and the difference is deliberate. An async connection pool (asyncpg above all) belongs to the event loop that first connected through it, while check() is driven synchronously from the application thread. If your application has a threaded or streaming gateway, that is a different loop from the one your request handlers run on — so a border that probed on its own would claim the pool for a loop no request could ever reach, and the application would fail every query while a perfectly healthy database was repeatedly "repaired". So the async border probes only when it can prove it is already on the pool's own loop, and otherwise reports a signal maintained by the real request path: a borrow that cannot reach the database marks the engine, a borrow that succeeds clears it.

The practical consequence, worth knowing: on the async road, detection follows traffic. An outage is noticed the moment a real request runs into it, and repair follows on the next tick — but an application sitting completely idle will not notice a database that died while it was idle, because nothing has asked. An application that never uses a threaded gateway (everything on one loop) keeps full active probing after its first borrow, since the border can then prove the probe is safe. AsyncSqlaEngine.is_alive() itself is unchanged and remains a real probe for any caller running on the pool's own loop.

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.

Give those variables a DSN with no +driver suffix:

export SQLA_TEST_POSTGRES_URI='postgresql://postgres:postgres@localhost:55432/postgres'

One variable feeds both roads, and each road appends the driver it needs -- asyncpg for the async engine, psycopg2 for the sync one. Pinning a driver in the DSN forces that one driver onto both, and the mismatched road fails at connect time with a keyword it does not understand (postgresql+asyncpg on the sync road: connect() got an unexpected keyword argument 'connect_timeout'). The sync-road extras are not installed by default, so a driverless DSN simply skips those tests rather than failing them.

License

MIT.