postgres-component

Async Postgres connection-lifecycle component for python-components.

Install

uv add postgres-component

Requirements

• Python >= 3.11
• A running Postgres server

Usage

import asyncio

from python_components import System
from postgres_component import PostgresComponent

database = PostgresComponent(url="postgresql+asyncpg://user:pass@localhost:5432/mydb")

system = System({"database": database})

async def main():
    async with system:
        async with database.session() as session:
            await session.execute(...)  # commits on success, rolls back on exception

asyncio.run(main())

PostgresComponent can also be built from individual kwargs instead of a URL:

PostgresComponent(host="localhost", port=5432, user="user", password="pass", database="mydb")

url takes priority over the kwargs when both are given.

Health route

from fastapi_component import create_app

app = create_app(system)  # RouteProvider discovery adds GET /health automatically

Migrations

PostgresComponent never runs migrations itself. migration_wiring() hands your own Alembic env.py the same connection config so it doesn't have to be re-derived:

# alembic/env.py
from postgres_component import create_migration_engine

engine = create_migration_engine(database.migration_wiring())
# use `engine` with Alembic's async run_sync() pattern

Semantics and caveats

• Fail-fast startup. start() runs an explicit SELECT 1 before considering itself connected — a dead database makes start() raise rather than "starting" successfully and failing later on first query.
• Managed Session is a unit of work. database.session() begins a transaction, commits on clean exit, and rolls back on exception — one call is one unit of work. For manual control (long-lived reads, streaming), use the raw database.session_factory directly.
• Errors propagate raw. SQLAlchemy exceptions (OperationalError, IntegrityError, etc.) are never wrapped; get_status()["last_error"] is updated as a string for introspection, but the original exception type always reaches the caller.
• No ORM, no migrations. The component owns connection lifecycle only. Models and Alembic setup belong to the consuming app; migration_wiring() only removes the boilerplate of re-deriving connection config.

Development

uv sync --all-groups
uv run pytest
uv run ruff format --check .
uv run ruff check .

Integration tests spin up a real Postgres instance via testcontainers. For local development against a persistent instance instead, run docker compose up -d.

License

MIT