FastAPI integration for dbwarden.
Provides request-scoped sessions, a migration-aware lifespan, and health/status routers.
| Hook | Provides |
|---|---|
session_factory / sync_session_factory |
Request-scoped AsyncSession / Session dependencies, backing database_config(...).async_session and .sync_session |
clickhouse_session_factory / clickhouse_sync_session_factory |
Shared ClickHouse client dependencies |
lifespan |
Startup schema check or auto-migration, optional readiness gate, seed application, and pool warmup; disposes engines on shutdown |
health_routes |
GET /, /liveness, /readiness, /{database_name} |
migration_routes |
GET /status and POST /migrate |
Also exported directly: dbwarden_lifespan, DBWardenRouter, DBWardenHealthRouter, MetricsRouter, MetricsMiddleware, QueryTracingMiddleware, PoolMetricsCollector, migration_lock, sync_migration_lock, override_database, and migration_state.
Re-exports schemap's @auto_schema and SchemaConfig, so Pydantic schemas can be generated from your SQLAlchemy models:
from dbwarden_fastapi import auto_schema
@auto_schema
class User(Base): ...from fastapi import FastAPI
from dbwarden_fastapi import DBWardenHealthRouter, dbwarden_lifespan
app = FastAPI(lifespan=lambda app: dbwarden_lifespan(app, mode="check"))
app.include_router(DBWardenHealthRouter(), prefix="/health")dbwarden plugin add dbwarden-fastapiOptional extras: [metrics] for Prometheus counters, [redis] for the distributed migration_lock, [clickhouse] for ClickHouse sessions.
get_tenant_session() selects a registered database for each request.
TenantResolver(source="header") reads X-DBWarden-Database by default. Use
source="host", source="both", precedence="host", or host_mapping to map
hostnames to registered database names. A custom sync or async callable accepting
Request may also be supplied. get_session() also accepts a registered
DatabaseHandle or DbwardenDatabase class.
For dbwarden_lifespan(mode="migrate"), background_migrations=True supports
background_migration_readiness="block" (default), "serve", or "fail".
Metrics refresh during lifespan every 30 seconds by default and refreshes stale
scrapes; set metrics_refresh_interval=None to disable the periodic task.
Set opentelemetry=True with the [opentelemetry] extra to instrument FastAPI.
This is an official dbwarden plugin. Its distribution name is classified before any of its code is imported, and dbwarden plugin add verifies the PyPI Trusted-Publishing attestation (PEP 740) against dbwarden-org/dbwarden-fastapi before installing. It loads automatically once installed, with no dbwarden plugin trust step.
uv venv && uv pip install -e . -e ../dbwarden pytest
pytest -qThe tests/test_conformance.py suite runs dbwarden's shared conformance harness (dbwarden.plugin_conformance): entry point resolution, no import-time side effects, hook signatures, public-API-only imports, and idempotent setup().
MIT