fastfoundry-postgres 1.0.1

PostgreSQL connection building blocks for the fastfoundry ecosystem.

fastfoundry-postgres

PostgreSQL connection building blocks for the fastfoundry ecosystem.

Part of the fastfoundry.* namespace (fastfoundry-settings, fastfoundry-redis, …).

An async SQLAlchemy engine and session lifecycle for a single PostgreSQL database (via asyncpg), a declarative model base, and a configuration model that plugs into fastfoundry.settings.

Install

pip install fastfoundry-postgres
# or
uv add fastfoundry-postgres

Requires Python 3.12+.

Configuration

DatabaseCfg holds the connection parameters. It is designed to nest inside a fastfoundry.settings.BaseAppSettings subclass, so the values load from the environment with the rest of your service configuration, then get handed to database_ctx:

from fastfoundry.settings import BaseAppSettings
from fastfoundry.postgres import DatabaseCfg


class Settings(BaseAppSettings):
    database: DatabaseCfg


settings = Settings()

With the FF_ prefix and __ nested delimiter that BaseAppSettings applies, the fields are read from the environment as FF_DATABASE__HOST, FF_DATABASE__PORT, FF_DATABASE__USER, and so on.

Engine & session lifecycle

Open database_ctx(settings.database) once for the application lifetime — it builds the engine and session factory, publishes them for ambient access (get_engine() and db_cfg()), and disposes the engine on exit. For FastAPI, wire it into the lifespan:

from contextlib import asynccontextmanager
from collections.abc import AsyncGenerator

from fastapi import FastAPI
from fastfoundry.postgres import database_ctx

from app.settings import settings


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    async with database_ctx(settings.database):
        yield


app = FastAPI(lifespan=lifespan)

Acquire a session with database_session, and open transactions explicitly with transaction (sessions use autobegin=False, so nothing starts a transaction for you — even read-only work should be wrapped, as SQLAlchemy recommends). A nested transaction becomes a SAVEPOINT:

from fastfoundry.postgres import database_session, transaction


async with database_session() as session:
    async with transaction(session):
        session.add(user)

Models

Subclass BaseDBModel. The table name defaults to the lower-cased class name with a trailing s; set __table_name__ to override it:

from sqlalchemy.orm import Mapped, mapped_column

from fastfoundry.postgres import BaseDBModel


class User(BaseDBModel):          # -> table "users"
    id: Mapped[int] = mapped_column(primary_key=True)


class Account(BaseDBModel):
    __table_name__ = "accounts"   # explicit table name
    id: Mapped[int] = mapped_column(primary_key=True)

Development

mise run install     # uv sync --dev
mise run lint        # ruff check
mise run typecheck   # mypy --strict
mise run test        # pytest
mise run build       # uv build (sdist + wheel)

License

MIT