Building Scalable APIs with FastAPI and PostgreSQL

FastAPI changed how I think about building Python web services. It's not just faster than Flask — it fundamentally shifts your mental model toward correctness first.

The Problem with Naive Async

Most tutorials show you this:

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    return user

And it works — until you hit 500 concurrent requests and your connection pool collapses.

Connection Pooling That Actually Works

The key is configuring asyncpg pool parameters to match your deployment environment:

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,
    pool_pre_ping=True,
)

pool_pre_ping=True is the one most people skip. It validates connections before checkout, catching stale connections before they ruin a user's request.

Structured Logging With Context

Add request ID propagation from the start — retrofitting it later is painful:

import structlog
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="")

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    rid = request.headers.get("X-Request-ID", str(uuid4()))
    request_id_var.set(rid)
    structlog.contextvars.bind_contextvars(request_id=rid)
    response = await call_next(request)
    response.headers["X-Request-ID"] = rid
    return response

Now every log line in that request's lifecycle carries the same request_id. Debugging production incidents becomes trivial.

What I Ship in Every New Service

1. Health check endpoint — /health with DB ping, returns structured JSON
2. Prometheus metrics — via prometheus-fastapi-instrumentator
3. Request ID middleware — as above
4. Graceful shutdown — handle SIGTERM to drain in-flight requests
5. Rate limiting — slowapi backed by Redis, not in-memory

None of these are optional in production. Build them in on day one.

The Pattern I Return To

class ServiceResult(Generic[T]):
    data: T | None
    error: str | None

    @classmethod
    def ok(cls, data: T) -> "ServiceResult[T]":
        return cls(data=data, error=None)

    @classmethod
    def fail(cls, error: str) -> "ServiceResult[T]":
        return cls(data=None, error=error)

This forces every service method to make the success/failure contract explicit. No more None returns that callers forget to check.

These patterns have survived real traffic spikes, on-call rotations, and teammates who skimmed the architecture docs. Start with them, not without.