Concepts¶

High-level explanations of how DBWarden's FastAPI integration works.

What Problem Does It Solve?¶

Without DBWarden, FastAPI apps typically have split configuration:

# migrations/env.py - Alembic config
SQLALCHEMY_DATABASE_URL = "postgresql://..."

# app/database.py - App config
SQLALCHEMY_DATABASE_URL = "postgresql://..."  # Duplicate!

# app/main.py - Manual engine creation
engine = create_async_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(engine)

# Manual dependency
async def get_db():
    async with SessionLocal() as session:
        yield session

With DBWarden, you have one source of truth:

# dbwarden.py - Single config
database_config(
    database_url="postgresql://...",
    model_paths=["app.models"],
)

# app/main.py - Automatic
SessionDep = Annotated[AsyncSession, Depends(get_session())]

DBWarden handles: - ✅ Engine creation - ✅ Session factories - ✅ Connection pooling - ✅ Startup checks - ✅ Health endpoints

Dependency Injection¶

FastAPI uses dependency injection to provide resources to routes.

Without DBWarden¶

# Manual dependency
async def get_db():
    async with SessionLocal() as session:
        try:
            yield session
        except:
            await session.rollback()
            raise

# Every route
@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    ...

With DBWarden¶

# One-time setup
SessionDep = Annotated[AsyncSession, Depends(get_session())]

# Every route
@app.get("/users")
async def list_users(session: SessionDep):
    ...

DBWarden's get_session() returns a dependency function that FastAPI calls automatically on each request.

Engine Caching¶

Why Cache Engines?¶

Creating database engines is expensive:

# ❌ Bad: New engine per request
@app.get("/users")
async def list_users():
    engine = create_async_engine(...)  # Expensive!
    async with engine.connect() as conn:
        ...

Engines should be created once and reused:

# ✅ Good: One engine for app lifetime
engine = create_async_engine(...)  # Created once

@app.get("/users")
async def list_users(session: SessionDep):
    # Reuses cached engine
    ...

DBWarden caches engines automatically:

First request:
1. get_session() called
2. Engine created
3. Engine cached
4. Session created from engine
5. Session yielded to route

Subsequent requests:
1. get_session() called
2. Engine retrieved from cache  ← Fast!
3. Session created from engine
4. Session yielded to route

Session Scope¶

Request-Scoped Sessions¶

Each request gets its own session:

Request 1: Session A
Request 2: Session B
Request 3: Session C

Sessions are never shared between requests, ensuring: - ✅ Transaction isolation - ✅ No race conditions - ✅ Predictable behavior

Session Lifecycle¶

Request arrives
    ↓
FastAPI calls get_session dependency
    ↓
New session created
    ↓
Session yielded to route
    ↓
Route executes with session
    ↓
Session automatically closed
    ↓
Response returned

If an error occurs, the session is rolled back before closing.

Health vs Liveness¶

Kubernetes has two types of probes:

Liveness Probe¶

Question: Is the app alive?

Answer: If no, restart the pod.

Use: Basic health check

livenessProbe:
  httpGet:
    path: /ping  # Simple endpoint, no DB
  failureThreshold: 3

Readiness Probe¶

Question: Is the app ready to serve traffic?

Answer: If no, stop routing traffic (but don't restart).

Use: Database health, migration state

readinessProbe:
  httpGet:
    path: /health/  # Full health check with DB
  failureThreshold: 2

DBWarden's health endpoints are perfect for readiness probes because they check: - Database connectivity - Migration state - Lock status

Async vs Sync¶

DBWarden's FastAPI integration is async-native.

Why Async?¶

Sync (blocking):

result = session.execute(select(User))  # Blocks thread
users = result.scalars().all()

While waiting for the database, the thread can't do anything else.

Async (non-blocking):

result = await session.execute(select(User))  # Releases control
users = result.scalars().all()

While waiting for the database, the event loop can handle other requests.

Result: Async can handle 10-100x more concurrent requests than sync with the same resources.

Async Drivers¶

DBWarden automatically uses async drivers:

Database	Async Driver
PostgreSQL	`asyncpg`
SQLite	`aiosqlite`

Your URLs are automatically upgraded:

# Your config
database_url="postgresql://localhost/myapp"

# DBWarden uses
database_url="postgresql+asyncpg://localhost/myapp"

expire_on_commit¶

This is a session setting that affects object behavior after commit.

Without expire_on_commit=False¶

user = User(email="test@example.com")
session.add(user)
await session.commit()

# ❌ Error: Instance is not bound to a Session
return user

After commit, SQLAlchemy expires all objects, meaning they're no longer accessible without a session.

With expire_on_commit=False¶

user = User(email="test@example.com")
session.add(user)
await session.commit()

# ✅ Works: Object still accessible
return user

Objects remain accessible after commit.

Why does DBWarden use this?

FastAPI serializes response objects after the route returns:

Route returns user
    ↓
Session closes
    ↓
FastAPI serializes user to JSON  ← Needs to access user.email!
    ↓
Response sent

Without expire_on_commit=False, serialization would fail because the session is closed.

Configuration Resolution¶

DBWarden resolves configuration in this order:

Explicit config - database_config(...) calls
Runtime flags - dev=True parameter
Environment variables - ENVIRONMENT=development
Default values - Built-in defaults

Example:

database_config(
    database_url="postgresql://prod-db/myapp",
    dev_database_url="sqlite:///dev.db",
)

# In development
get_session(dev=True)  # Uses sqlite:///dev.db

# In production
get_session()  # Uses postgresql://prod-db/myapp

When to Use DBWarden¶

Use DBWarden when: - ✅ You want migrations and runtime to share config - ✅ You need startup validation - ✅ You want built-in health endpoints - ✅ You're building a new FastAPI app - ✅ You use SQLAlchemy for models

Don't use DBWarden when: - ❌ You don't use SQLAlchemy - ❌ You already have working migration infrastructure - ❌ You use an ORM other than SQLAlchemy (e.g., Tortoise, SQLModel standalone)

Comparison to Alternatives¶

vs. Alembic + Manual Setup¶

	DBWarden	Alembic + Manual
Configuration	One source	Split (env.py + app code)
Engine creation	Automatic	Manual
Session dependency	Built-in	Custom
Startup checks	Built-in	Custom
Health endpoints	Built-in	Custom
Learning curve	Lower	Higher

vs. SQLModel¶

SQLModel includes SQLAlchemy but doesn't provide: - Migration management - Startup checks - Health endpoints - Multi-database support

DBWarden can work with SQLModel for migrations while you use SQLModel's ORM.

vs. Django ORM¶

Django's ORM is integrated with Django's migration system. DBWarden is for FastAPI + SQLAlchemy apps.

Recap¶

✅ DBWarden provides one configuration source for migrations and runtime
✅ Uses dependency injection for clean, reusable code
✅ Caches engines for performance
✅ Sessions are request-scoped and automatically managed
✅ Health endpoints integrate with Kubernetes probes
✅ Async-native for high concurrency
✅ expire_on_commit=False for FastAPI compatibility
✅ Configuration resolves from explicit → environment → defaults

What's Next?¶

API Reference - Complete function signatures
Tutorial - Build your first app