9. FastAPI Integration¶

What You'll Learn¶

• How to wire DBWarden into a FastAPI application lifecycle
• How to use primary.async_session as a dependency injection
• How to expose health check and migration endpoints
• How to validate schema on startup

Prerequisites¶

• Docker (for PostgreSQL)
• examples/fastapi-app/ directory

Step 1: Configuration with Session Handles¶

from dbwarden import database_config

primary = database_config(
    database_name="primary",
    default=True,
    database_type="postgresql",
    database_url_sync="postgresql://user:password@localhost:5432/myapp",
    database_url_async="postgresql+asyncpg://user:password@localhost:5432/myapp",
    model_paths=["app.models"],
)

The primary object is a DatabaseHandle. It exposes primary.async_session and primary.sync_session as FastAPI-compatible dependency annotations; no separate dependency module needed.

Step 2: Lifespan Hook¶

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)

dbwarden_lifespan runs on every startup:

1. Schema validation (mode "check"): verifies all pending migrations exist and the database is in a known state
2. Readiness gate: the app won't accept traffic until validation passes
3. Connection pool warmup: pre-connects to the database
4. On shutdown: disposes all engine pools and ClickHouse clients

Available modes: - "check": validate schema, fail on pending migrations (recommended for production) - "migrate": apply pending migrations automatically on startup - "skip": no startup checks

Step 3: Session Dependency in Routes¶

from config import primary
from app.models import User
from app.schemas import UserResponse


@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)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

primary.async_session is a type alias for Annotated[AsyncSession, Depends(...)]. FastAPI resolves it to an actual database session using the engine configured in database_config().

The session is automatically: - Opened when the route handler starts - Committed (or rolled back on exception) when the handler finishes - Closed and returned to the pool

Step 4: Health Endpoints¶

from dbwarden.fastapi import DBWardenHealthRouter

app.include_router(DBWardenHealthRouter(), prefix="/health")

This adds:

Sample response:

{
  "status": "ok",
  "databases": {
    "primary": {
      "status": "ok",
      "connected": true,
      "pending_migrations": 0,
      "applied_migrations": 5,
      "lock_active": false
    }
  }
}

Step 5: Migration Endpoints¶

from dbwarden.fastapi import DBWardenRouter

app.include_router(DBWardenRouter(), prefix="/db")

These endpoints are useful for management UIs or automated deployment tooling.

Step 6: The Complete App¶

from contextlib import asynccontextmanager
from fastapi import FastAPI
from dbwarden.fastapi import (
    DBWardenHealthRouter,
    DBWardenRouter,
    dbwarden_lifespan,
)
from app.routes import users


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


app = FastAPI(
    title="DBWarden FastAPI Example",
    lifespan=lifespan,
)

app.include_router(users.router, prefix="/api/v1")
app.include_router(DBWardenHealthRouter(), prefix="/health")
app.include_router(DBWardenRouter(), prefix="/db")

Step 7: Run and Test¶

# Install dependencies
uv add dbwarden sqlalchemy fastapi uvicorn asyncpg

# Start PostgreSQL
docker run -d --name pg -e POSTGRES_USER=user \
    -e POSTGRES_PASSWORD=password -e POSTGRES_DB=myapp \
    -p 5432:5432 postgres:16

# Initialize and migrate
$ dbwarden init
$ dbwarden make-migrations "create users table"
$ dbwarden migrate

# Start the app
uvicorn app.main:app --reload

# Health check
curl http://localhost:8000/health/

# Create a user
curl -X POST http://localhost:8000/api/v1/users/ \
    -H "Content-Type: application/json" \
    -d '{"email": "alice@example.com", "username": "alice"}'

# Migration status
curl http://localhost:8000/db/status

Key Takeaways¶

• database_config() returns a DatabaseHandle with built-in FastAPI dependencies
• dbwarden_lifespan integrates schema validation into the app lifecycle
• primary.async_session works directly as a route parameter type annotation
• DBWardenHealthRouter exposes liveness, readiness, and per-database health
• DBWardenRouter exposes migration status and execution as HTTP endpoints

Related Documentation¶

• FastAPI Integration Overview
• FastAPI Tutorial: First Steps
• FastAPI Tutorial: Complete Application
• Session Dependency
• Health Endpoints

Next¶

Section 10: Auto Schemas