FastAPI dependency injection and the request lifecycle

A handler needs a database session, an HTTP client, the current user. Three ways to give it them:

# A. Module-level global — easy, untestable, leaks state across requests.
db = SessionLocal()

# B. Construct inside the handler — couples the handler to the construction.
@app.post("/users")
def create_user(payload):
    db = SessionLocal()
    ...

# C. Inject — the handler declares dependencies, the framework supplies them.
@app.post("/users")
def create_user(payload, db: Session = Depends(get_db)):
    ...

Approach C is dependency injection. The handler signature is the contract; the framework walks that signature, finds matching providers, and supplies them at call time. The provider for db can be swapped (test fake, alternate engine, read replica) without touching the handler.

This pattern is older than FastAPI — Spring, Angular, and Django's class-based view system all do it. FastAPI's twist is that providers are themselves plain callables (including async ones, including generators for cleanup), so the same machinery handles auth checks, DB sessions, and request-scoped caches.

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)) -> dict:
    return db.query(User).get(user_id).as_dict()

What FastAPI did on the way in:

1. Read read_user's signature: user_id: int (path), db: Session = Depends(get_db) (dependency).
2. Resolved the path parameter from the URL, validated as int.
3. Called get_db(), drove the generator to the yield, captured the db value.
4. Called read_user(user_id=42, db=<Session>).
5. After building the response, resumed the generator past the yield so db.close() runs.

If get_db itself had a Depends(...) parameter, FastAPI would resolve it first — recursively, with cycle detection.

Dependencies compose. Auth checks are usually the first layer:

def get_token(request: Request) -> str:
    token = request.headers.get("authorization", "").removeprefix("Bearer ")
    if not token:
        raise HTTPException(401)
    return token

def get_current_user(
    token: str = Depends(get_token),
    db: Session = Depends(get_db),
) -> User:
    user = db.query(User).filter_by(token=token).first()
    if not user:
        raise HTTPException(401)
    return user

@app.get("/me")
def me(user: User = Depends(get_current_user)) -> dict:
    return user.as_dict()

Important: FastAPI caches each dependency result per request. If three handlers all Depends(get_db), the same Session is reused — not three sessions. This defeats the diamond problem (two paths to the same dep producing two instances) and is also why DB sessions and auth checks fit cleanly into Depends.

Pass use_cache=False when you really do want a fresh value per call site — e.g. a request-scoped UUID factory.

Two boundaries, both crossed by every request:

• In — query, path, and body parameters are validated against the handler's annotations before the handler runs. Invalid input → HTTP 422 with a structured error. Handler code can assume types are correct.
• Out — return values are serialized through the route's response_model (or the return-type annotation). Extra fields are dropped, sensitive fields filtered, dates ISO-formatted.

from pydantic import BaseModel, EmailStr, Field

class CreateUser(BaseModel):
    email: EmailStr
    nickname: str = Field(min_length=3, max_length=32)

class UserOut(BaseModel):
    id: int
    email: EmailStr
    nickname: str
    model_config = {"from_attributes": True}

@app.post("/users", response_model=UserOut)
def create_user(payload: CreateUser, db: Session = Depends(get_db)) -> UserOut:
    u = User(email=payload.email, nickname=payload.nickname)
    db.add(u); db.commit(); db.refresh(u)
    return u   # SQLAlchemy row → UserOut via from_attributes

Pydantic v2's validator is compiled in Rust (pydantic-core). The cost is paid once at model-class creation, then amortized per request. Throughput on FastAPI mostly tracks handler + DB, not validation.

The yield form of a dependency is a context manager in disguise. The framework runs setup before the handler, suspends at yield, resumes after the response:

def db_transaction(db: Session = Depends(get_db)):
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise

Now any handler that takes Depends(db_transaction) gets a transactional session: commit on success, rollback on any unhandled exception. If db.commit() itself raises (constraint violation, deadlock), the framework propagates it through the dependency, returning a 500 — or whatever exception handler is registered.

The try / except / raise shape matters: swallowing the exception silently would commit a partial transaction and return success to the client. Always re-raise; map to a user-facing status code at the exception-handler layer with @app.exception_handler(IntegrityError).

FastAPI's BackgroundTasks runs after the response is sent. It is in-process and lost on crash:

from fastapi import BackgroundTasks

@app.post("/orders")
def create_order(payload: OrderIn, bg: BackgroundTasks):
    order = save_order(payload)
    bg.add_task(send_receipt_email, order.id)
    return {"id": order.id}

Use it when the side-effect is best-effort, idempotent on retry, and small. Trade-offs:

BackgroundTasks runs in the same process and same event loop as the handler — a long task blocks shutdown and is lost if the worker is OOM-killed. Anything stateful, slow, or business-critical needs a real broker behind it.

Per-request state belongs in dependencies. Process-lifetime state — DB connection pools, HTTP clients, model weights — belongs in the ASGI lifespan context:

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, Request
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=5.0)
    app.state.engine = create_async_engine(DATABASE_URL)
    yield
    await app.state.http.aclose()
    await app.state.engine.dispose()

app = FastAPI(lifespan=lifespan)

def get_http(request: Request) -> httpx.AsyncClient:
    return request.app.state.http

@app.get("/proxy")
async def proxy(url: str, http: httpx.AsyncClient = Depends(get_http)):
    return (await http.get(url)).json()

Connection pools are expensive to build. A per-request httpx.AsyncClient() constructs and tears down a pool on every call, which dominates latency for low-CPU handlers. Anchor pools at lifespan, expose them through a thin dependency, and the per-request cost collapses to an attribute read.

The single biggest practical win of DI: app.dependency_overrides.

from fastapi.testclient import TestClient

def fake_db():
    yield InMemoryDB()

def fake_user():
    return User(id=1, email="t@example.com")

app.dependency_overrides[get_db] = fake_db
app.dependency_overrides[get_current_user] = fake_user

client = TestClient(app)
resp = client.get("/me")
assert resp.status_code == 200
assert resp.json()["email"] == "t@example.com"

Every handler that depended on get_db or get_current_user now gets the fake — without monkey-patching, without touching the handlers, without an import-time side effect.

The pattern: keep dependencies thin and named — one function per resource. Test overrides target exactly what they need to fake. The same trick removes auth in tests (get_current_user returns a fixed user), the same trick injects a deterministic clock (get_now), the same trick stubs the payment gateway (get_stripe).