FastAPI gets you to a working API in an afternoon, which is exactly why so many of them reach production half-dressed. Nothing in the framework forces you to think about connection pools, correlation IDs, or what happens when the container gets a SIGTERM mid-request. This is the checklist we walk before a FastAPI service goes live.
Settings: one object, validated at boot
Scattered os.getenv("STRIPE_KEY") calls are how you find out at 2am that a worker started fine and then blew up on the first webhook. One Settings class, validated at import, is the fix.
# app/config.py
from functools import lru_cache
from pydantic import PostgresDsn, RedisDsn, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="forbid")
environment: str = "production"
database_url: PostgresDsn
redis_url: RedisDsn
stripe_secret_key: str = Field(min_length=20)
jwt_secret: str = Field(min_length=32)
cors_origins: list[str] = []
db_pool_size: int = 5
db_max_overflow: int = 5
@lru_cache
def get_settings() -> Settings:
return Settings()
extra="forbid" catches typos in env var names, which otherwise fail silently. The @lru_cache accessor gives one instance per process and doubles as a DI provider you can override in tests. We call get_settings() during startup so a missing JWT_SECRET kills the container immediately instead of surfacing as a 500 on the first login.
Dependency injection is your test seam
Depends is not decoration. It's the cleanest way to swap the database, the current user, or a feature flag lookup in tests without monkeypatching imports.
async def get_db() -> AsyncIterator[AsyncSession]:
async with SessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
In tests, app.dependency_overrides[get_db] = _test_db swaps in a session bound to a transaction rolled back after each case. Same for get_current_user, so you never mint real JWTs in unit tests. Anything a route reaches for that isn't pure computation should arrive through Depends.
Async sessions and pool math
SQLAlchemy 2.0 async with async_sessionmaker, one session per request, commit on success and rollback on exception. The lifespan owns the engine:
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
@asynccontextmanager
async def lifespan(app: FastAPI):
s = get_settings()
engine = create_async_engine(
str(s.database_url),
pool_size=s.db_pool_size,
max_overflow=s.db_max_overflow,
pool_pre_ping=True,
pool_recycle=1800,
)
app.state.engine = engine
app.state.sessionmaker = async_sessionmaker(engine, expire_on_commit=False)
yield
await engine.dispose()
app = FastAPI(lifespan=lifespan, title="Billing API", version="1.0.0")
Do the pool arithmetic before deploy. Peak connections is workers x (pool_size + max_overflow). Four Uvicorn workers at 5 + 5 is 40 connections from one container. Add a second container, a migration job, and a worker fleet, against a managed Postgres with max_connections = 100, and you're one autoscale event from FATAL: sorry, too many clients already. We size for peak, leave 20% headroom for admin sessions, and put PgBouncer in transaction mode in front of anything that scales horizontally.
The other classic: a blocking call inside an async def handler. One requests.post() to Stripe, or a Pillow resize, and the event loop for that worker stalls while every other in-flight request waits. Use an async client (httpx.AsyncClient), push the work to run_in_threadpool, or define the route as plain def and let Starlette run it in the threadpool. That last option is underrated.
Background work
BackgroundTasks runs after the response, in the same process, with no persistence and no retries. If the pod restarts, the work is gone. Fine for things you can afford to lose: a cache warm, a non-critical audit write.
Anything a client would notice going missing needs a real queue. We reach for arq when Redis is already in the stack and the job graph is simple, Celery when there are beat schedules, chains, and priority queues. Three rules regardless of tool:
- Jobs are idempotent. Pass an ID, not an object graph, and make the handler safe to run twice. At-least-once delivery means it will run twice.
- Retries use exponential backoff with jitter. Five attempts at 2s, 8s, 30s, 2m, 10m covers most transient failures without hammering a downstream that's already struggling.
- Exhausted jobs go to a dead-letter queue with the original payload and the last traceback, and something alerts when that queue is non-empty. A DLQ nobody watches is a delete.
Structured logging and request IDs
JSON logs to stdout, one line per event, structlog doing the formatting. Every request gets an ID from the inbound X-Request-ID header or a fresh UUID, bound into a ContextVar so it appears on every log line downstream without being threaded through function signatures.
import time, uuid, structlog
from starlette.middleware.base import BaseHTTPMiddleware
log = structlog.get_logger()
class RequestContextMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
rid = request.headers.get("x-request-id") or str(uuid.uuid4())
structlog.contextvars.bind_contextvars(request_id=rid)
start = time.perf_counter()
try:
response = await call_next(request)
finally:
structlog.contextvars.clear_contextvars()
duration_ms = round((time.perf_counter() - start) * 1000, 2)
log.info(
"http_request",
method=request.method,
path=request.url.path,
status=response.status_code,
duration_ms=duration_ms,
)
response.headers["x-request-id"] = rid
return response
Propagate that header on outbound calls so a trace survives across services. Never log tokens, API keys, card data, full request bodies, or plaintext email addresses. To correlate a user, log the user ID. INFO for request lines and state changes, WARNING for handled degradation, ERROR only when a human should look. Debug logs left on in production are how a duration_ms field ends up buried under 40 lines of SQL echo.
Health checks
Two endpoints, different jobs. /healthz returns 200 as long as the process is alive and does nothing else. It must not touch the database: a brief Postgres blip would otherwise make Kubernetes restart every healthy pod at once, turning a 30-second incident into a 10-minute one.
/readyz checks dependencies and controls traffic:
async def _probe(name: str, coro_fn, seconds: float) -> tuple[str, str]:
try:
async with asyncio.timeout(seconds):
await coro_fn()
return name, "ok"
except Exception:
return name, "fail"
@router.get("/readyz", include_in_schema=False)
async def readyz(db: AsyncSession = Depends(get_db), redis: Redis = Depends(get_redis)):
results = await asyncio.gather(
_probe("db", lambda: db.execute(text("SELECT 1")), 2.0),
_probe("cache", redis.ping, 1.0),
)
checks = dict(results)
ok = all(v == "ok" for v in checks.values())
return JSONResponse({"checks": checks}, status_code=200 if ok else 503)
Timeouts are the point. A readiness probe without one hangs for the full probe timeout and tells you nothing.
OpenAPI hygiene
The schema is a contract, so treat it like one. Every route declares response_model and an explicit status_code (201 on create, 204 on delete). Routes are grouped under tags and mounted at /api/v1. Errors use a single envelope model registered via responses={404: {"model": ErrorResponse}}, so clients parse one shape instead of guessing. Bare HTTPException(400, "bad") strings leak into the frontend as untyped noise.
Once the schema is clean, generate the frontend client from it instead of hand-writing fetch wrappers. openapi-typescript in CI, with the build failing when committed types drift from the live schema. A renamed field then breaks the frontend build rather than a user's checkout.
The last mile
- Gunicorn with
uvicorn.workers.UvicornWorker, workers roughly2 x cores, lower if handlers are memory-heavy. Set--graceful-timeout 30so in-flight requests finish during a rolling deploy. - Graceful shutdown in the lifespan: dispose the engine, close the Redis pool, drain the HTTP client.
- Timeouts at every layer: a server-side ceiling,
httpx.Timeout(connect=3, read=10)on outbound calls,statement_timeouton the database role. - CORS listing exact origins.
allow_origins=["*"]withallow_credentials=Trueis ignored by browsers anyway. - Rate limiting on auth and anything expensive, keyed by user ID where possible, IP as fallback.
How we work
We run this checklist before every FastAPI handover, and the answers go into the repo rather than a conversation. That means a .env.example matching the Settings model field for field, the pool math written next to the deploy config, a runbook for the DLQ, and probe paths already wired into the platform's health checks. Clients usually inherit these services with a small internal team or a maintenance retainer, so the goal is that the next engineer can work out why something is set the way it is without asking us.