af-db-session

A configurable PostgreSQL session factory for Python applications. Wraps SQLAlchemy connection pool setup and pydantic-settings configuration into a single reusable package — install it, point it at your .env, and get sessions.

Installation

pip install af-db-session
# or with Poetry:
poetry add af-db-session

A PostgreSQL driver is not included — install whichever you prefer alongside it:

pip install psycopg2-binary   # most common
pip install psycopg           # psycopg3

Quickstart

from allfly.db.session import DatabaseSettings, build_session_factory

settings = DatabaseSettings()           # reads DB_* vars from .env
factory = build_session_factory(settings)

session = factory.get_session()
try:
    result = session.execute(...)
    session.commit()
finally:
    session.close()

# On app shutdown
factory.close()

Configuration

All settings are loaded from environment variables with a DB_ prefix. By default the library reads from a .env file in the working directory.

Environment variables

Choosing your env file

# Standard .env (default)
settings = DatabaseSettings()

# Custom env file — e.g. .env.local, .env.production
settings = DatabaseSettings.from_env(".env.local")

# No file — reads only from real environment variables
settings = DatabaseSettings.from_env(None)

# No file, with inline overrides
settings = DatabaseSettings.from_env(None, host="db.internal", database="myapp")

Using a different driver

# psycopg3
settings = DatabaseSettings.from_env(".env", driver="psycopg")

# or via env var
# DB_DRIVER=psycopg

The driver value is used as the SQLAlchemy URL scheme: postgresql+{driver}://.... The corresponding package must be installed in your environment.

Read-only replica

from allfly.db.session import DatabaseSettings, build_ro_session_factory

settings = DatabaseSettings()       # set DB_RO_HOST to point at your replica
ro_factory = build_ro_session_factory(settings)

session = ro_factory.get_session()  # writes will be rejected by PostgreSQL

If DB_RO_HOST is not set, RODatabaseSessionFactory falls back to the primary host but still enforces read-only mode at the PostgreSQL level.

FastAPI example

from contextlib import asynccontextmanager
from fastapi import FastAPI
from allfly.db.session import DatabaseSettings, build_session_factory

settings = DatabaseSettings.from_env(".env.local")

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db = build_session_factory(settings)
    yield
    app.state.db.close()

app = FastAPI(lifespan=lifespan)

@app.get("/items")
def list_items():
    session = app.state.db.get_session()
    try:
        return session.execute(...).all()
    finally:
        session.close()

Logging

The library emits to the allfly.db.session logger namespace using Python's standard logging module. To activate debug output:

import logging
logging.getLogger("allfly.db.session").setLevel(logging.DEBUG)

Routing to loguru

If your app uses loguru, intercept stdlib logging once at startup:

import logging
from loguru import logger

class InterceptHandler(logging.Handler):
    def emit(self, record: logging.LogRecord) -> None:
        logger.opt(depth=6, exception=record.exc_info).log(
            record.levelname, record.getMessage()
        )

logging.getLogger("allfly.db.session").addHandler(InterceptHandler())