>
dbwarden
GitHub ↗
/ fastapi

Database Migrations for
FastAPI + SQLAlchemy.

Using Alembic with FastAPI? dbwarden is a declarative alternative for SQLAlchemy applications. The dbwarden-fastapi plugin provides sessions, health checks, and migrations from the same dbwarden.py file that generates your migrations.

/ 01 One configRead the docs ↗

The database object is the integration point.

The plugin reads the same DbwardenDatabase objects you already configure for migrations, so FastAPI never needs a second source of truth for connection settings. Primary.handle is a DatabaseHandle: it carries the sync engine used by migrate and the async engine used by your routes, and it exposes both as dependency annotations. The database_config(...) function API returns the same handle, so existing function-style configs get the same integration without a rewrite.

Both URLs are declared on the same object: database_url_sync for the migration tooling and database_url_async for FastAPI. The async URL conventionally uses postgresql+asyncpg://; the plugin builds the pool from that engine when the app starts.

the config
from dbwarden import DbwardenDatabase

class Primary(DbwardenDatabase):
    database_name = "primary"
    default = True
    database_type = "postgresql"
    database_url_sync = "postgresql://user:pass@localhost:5432/myapp"
    database_url_async = "postgresql+asyncpg://user:pass@localhost:5432/myapp"
    model_paths = ["app.models"]
/ 02 StartupRead the docs ↗

Startup checks the schema, or applies pending migrations.

dbwarden_lifespan runs on every startup and does four things: it validates the schema, it acts as a readiness gate so the app does not accept traffic until validation passes, it warms up the connection pool, and on shutdown it disposes every engine pool and ClickHouse client. The lifespan is entered as an async context manager, so the cleanup runs even when the app exits abnormally.

The mode decides what validation means. check verifies that no pending migrations exist and fails startup otherwise, which is the production recommendation: an out-of-date database never takes traffic. migrate applies pending migrations automatically before the app serves, for environments that should self-apply. skip runs no startup checks at all, for local debugging or when migrations are managed elsewhere.

the lifespan
from contextlib import asynccontextmanager
from fastapi import FastAPI
from dbwarden_fastapi import dbwarden_lifespan

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with dbwarden_lifespan(app, mode="check"):
        yield

app = FastAPI(lifespan=lifespan)
/ 03 SessionsRead the docs ↗

A session dependency from the config.

primary.async_session is a type alias for Annotated[AsyncSession, Depends(...)], generated from the config. FastAPI resolves it to a real session backed by the engine configured on Primary, so route handlers never construct sessions or pick an engine themselves.

The session lifecycle is handled for you: it is opened when the handler starts, committed when the handler returns, rolled back if the handler raises, and closed and returned to the pool either way. Because the dependency resolves through the request, the same annotation works for sync handlers via sync_session, and ClickHouse clients get their own session factories from the plugin's ClickHouse extra.

the route
from config import primary

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, session: primary.async_session):
    result = await session.execute(
        select(User).where(User.id == user_id)
    )
    return result.scalar_one_or_none()
/ 04 API schemasRead the docs ↗

Request and response models from the models.

@auto_schema turns a model into four schema classes: CreateSchema for POST bodies, UpdateSchema for PATCH, PublicSchema for responses, and Schema for internal use. Field definitions and validation come from the model columns, so a schema change and a migration change land in the same commit.

Fields marked public = False in the model's Meta are excluded from PublicSchema, which is where password hashes and internal flags belong. The generated schemas are Pydantic models, so they work with FastAPI's validation, OpenAPI generation, and model_validate for turning ORM instances into responses.

The four classes differ in what they require: CreateSchema expects the fields needed to create a row, UpdateSchema makes them optional for partial updates, and Schema is the internal representation used between layers. Validation failures surface as FastAPI's standard 422 responses, and every schema appears in the generated OpenAPI docs, so the API contract is derived from the models rather than maintained by hand.

the schemas
@auto_schema
class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)

    class Meta(TableMeta):
        class password_hash:
            public = False  # excluded from PublicSchema

create = User.CreateSchema(email="[email protected]", password_hash="secret")
public = User.PublicSchema.model_validate(user)
/ 05 Health and operationsRead the docs ↗

Health and migration endpoints over HTTP.

Two routers expose the operational surface. The health router reports per-database status: connectivity, applied and pending migrations, and whether the migration lock is active. GET /health/readiness checks the database before traffic is routed to the instance, which is what makes it useful as a Kubernetes readiness probe, and GET /health/liveness stays lightweight for the liveness probe.

The migration router exposes GET /db/status, a JSON representation of dbwarden status, and POST /db/migrate, which applies pending migrations at runtime. Those endpoints are for management UIs and automated deployment tooling. Both routers can be protected: DBWARDEN_MIGRATE_AUTH requires an X-API-Key header on the migrate endpoint, and DBWARDEN_HEALTH_AUTH does the same for health.

the routers
from dbwarden_fastapi import DBWardenHealthRouter, DBWardenRouter

app.include_router(DBWardenHealthRouter(), prefix="/health")
app.include_router(DBWardenRouter(), prefix="/db")
/ 06 Metrics and logsRead the docs ↗

Prometheus metrics and JSON logs.

With DBWARDEN_METRICS=true, migrate and seed commands record counters, gauges, and histograms: migrations applied, migration errors, schema and seed version, pending migrations, durations, and errors, all labeled by database. The plugin exposes them at /metrics through the MetricsRouter and MetricsMiddleware, so a single scrape target covers both the app and the migration commands it runs.

DBWARDEN_LOG_JSON switches all logs to newline-delimited JSON for ELK, Loki, or Datadog. When you need to see exactly what ran against the database, --debug-level trace logs every SQL statement as it executes, and --perf adds per-statement timing. The metrics dependency is optional: uv add "dbwarden[metrics]", and without it every metric function is a safe no-op.

metrics and json
$ DBWARDEN_METRICS=true dbwarden migrate
$ DBWARDEN_LOG_JSON=true dbwarden migrate --debug-level trace
/ fastapi + sqlalchemy + postgresql

The strongest combination for most projects is FastAPI + SQLAlchemy + PostgreSQL. dbwarden's PostgreSQL backend provides full round-trip support: identity columns, generated columns, partitioning, row-level security, exclusion constraints, advanced indexes, enums, domains, composite types, sequences, functions, triggers, roles, and event triggers all round-trip from models to SQL and back.

The FastAPI plugin adds async session dependencies backed by postgresql+asyncpg://, health endpoints for Kubernetes probes, and Prometheus metrics: all wired to the same DbwardenDatabase config that drives migrations. Your PostgreSQL-specific options (fill factor, schema, identity) live in typed metadata on the models, not in a separate HCL or SQL file.

from dbwarden import DbwardenDatabase

class Primary(DbwardenDatabase):
    database_name = "primary"
    default = True
    database_type = "postgresql"
    database_url_sync = "postgresql://user:pass@localhost:5432/myapp"
    database_url_async = "postgresql+asyncpg://user:pass@localhost:5432/myapp"
    model_paths = ["app.models"]

Potential long-tail queries this page captures: FastAPI PostgreSQL migrations, FastAPI SQLAlchemy PostgreSQL, FastAPI Alembic alternative, SQLAlchemy PostgreSQL migration tool.

/ faq
Does FastAPI support require the plugin?

Yes. FastAPI integration ships as the official dbwarden-fastapi plugin, installed with dbwarden plugin add dbwarden-fastapi. Core stays framework-independent.

Is there a sync session dependency too?

Yes: Primary.handle exposes both async_session and sync_session as FastAPI-compatible dependency annotations. ClickHouse clients get their own session factories.

Which lifespan mode should production use?

check. The app validates the schema and fails startup on pending migrations, so an out-of-date database never takes traffic. Use migrate for environments that should self-apply.

Can the migrate endpoint be protected?

Yes. DBWARDEN_MIGRATE_AUTH requires an X-API-Key header on POST /db/migrate, and DBWARDEN_HEALTH_AUTH does the same for the health endpoints.

Do the metrics require an extra install?

The plugin ships optional extras: [metrics] for Prometheus recording, [redis] for the distributed migration lock, and [clickhouse] for ClickHouse sessions.

/ where it lives
An official plugin

FastAPI support ships as dbwarden-fastapi, installed with dbwarden plugin add dbwarden-fastapi. Core stays framework-independent.

The full reference

Session dependencies, health routes, metrics, and @auto_schema live in the plugin repository and its docs.

dbwarden-fastapi on GitHub ↗