fastfoundry-postgres 1.1.0

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 = DatabaseCfg()


settings = Settings()

Give it either a single dsn or the discrete fields. With the FF_ prefix and __ nested delimiter that BaseAppSettings applies, that is one of:

# a single DSN secret …
FF_DATABASE__DSN=postgresql+asyncpg://user:pass@host:5432/db

# … or discrete fields
FF_DATABASE__HOST=host
FF_DATABASE__PORT=5432
FF_DATABASE__USER=user
FF_DATABASE__PASSWORD=pass
FF_DATABASE__DATABASE=db

Or construct it directly — DatabaseCfg(dsn="postgresql+asyncpg://user:pass@host:5432/db") or DatabaseCfg(host="host", user="user", password="pass", database="db"). Notes:

• When dsn is given it is authoritative: its host/port/user/password/database overwrite the discrete fields. Pool sizing (min_size / max_size / command_timeout) is never part of a DSN and always comes from the fields (defaults 5 / 10 / 60).
• dsn and password are SecretStr, so they are masked in repr and never leak when the settings object is logged.
• The scheme/driver in the DSN is parsed but not retained — database_ctx always connects via postgresql+asyncpg. The port defaults to 5432 when the DSN omits it.
• The discrete fields carry sensible PostgreSQL defaults (localhost / 5432 / postgres / "" / postgres), so the config is always fully populated and typed.

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