FastAPI

The FastAPI helpers are thin, read-only wrappers around client.info(). They report pending migrations for a lifespan guard or a health endpoint, and never apply anything themselves.

extra · dblift[fastapi]

Install

pip install "dblift[fastapi]"

For PostgreSQL, include the driver extra: pip install "dblift[postgresql,fastapi]".

Keep asyncpg for application queries

DBLift can run migrations as a separate deploy step. Your FastAPI routes and workers can keep using asyncpg; they do not need a SQLAlchemy connection.

Put versioned SQL files such as migrations/V1__create_jobs.sql in your project, then run:

export DBLIFT_DB_URL="postgresql+psycopg://user:password@localhost:5432/app"
dblift migrate --dry-run --show-sql
dblift migrate
dblift validate

The DBLift CLI uses its own PostgreSQL driver connection. The optional startup guard below uses a separate SQLAlchemy/psycopg engine to check for pending migrations; it does not use or replace your application's asyncpg pool, and it never applies migrations.

Startup guard

main.py

python

from contextlib import asynccontextmanager

from fastapi import FastAPI
from sqlalchemy import create_engine

from dblift.api import DBLiftClient
from dblift.integrations.fastapi import migration_guard

engine = create_engine("postgresql+psycopg://user:password@localhost/app")

@asynccontextmanager
async def lifespan(app: FastAPI):
    client = DBLiftClient.from_sqlalchemy(engine=engine, migrations_dir="migrations")
    try:
        migration_guard(client, on_pending="raise")
        app.state.dblift_client = client
        yield
    finally:
        client.close()

app = FastAPI(lifespan=lifespan)

on_pending	Behaviour when migrations are pending
`"raise"`	Raises, listing the pending items. The default, and the usual choice in a lifespan.
`"warn"`	Emits a warning through the warnings module and continues.
`"ignore"`	Does nothing — no info() call at all.

Health endpoints

main.py

python

from dblift.integrations.fastapi import check_migrations_current, health_payload

@app.get("/health")
def health():
    return health_payload(app.state.dblift_client)

@app.get("/migrations/pending")
def pending():
    return {
        "pending": check_migrations_current(app.state.dblift_client),
    }

health_payload returns pending_migrations, current, current_schema_version and pending_count. check_migrations_current returns the pending identifiers alone, and an empty list when the database is current.

Async mirrors

main.py

python

from dblift.api.async_client import AsyncDBLiftClient
from dblift.integrations.fastapi import (
    check_migrations_current_async,
    health_payload_async,
    migration_guard_async,
)

async with AsyncDBLiftClient.from_sqlalchemy(
    engine,
    migrations_dir="migrations",
) as client:
    await migration_guard_async(client, on_pending="raise")
    payload = await health_payload_async(client)

Applying stays explicit

None of these helpers call migrate(). Apply from a deploy step, an admin route, or the CLI.

FastAPI

Install

Keep asyncpg for application queries

Startup guard

Health endpoints

Async mirrors

Read next