FastAPI in depth — routers, dependencies, and lifespan — Python Mastery

Why this matters

FastAPI is the dominant Python web framework in 2026 — async by default, type-driven, auto-generated OpenAPI docs, used everywhere from agent serving to enterprise APIs. This lesson covers the patterns that take you from "hello world" to a production-ready service: dependencies, middleware, background tasks, lifespan, auth, error handling, testing, deployment.

Learning objectives

Build async endpoints with Pydantic models.
Use dependency injection with Depends.
Use lifespan / middleware / background tasks.
Implement auth, error handling, validation.
Test with httpx.AsyncClient.

1. Hello FastAPI

powershell

uv add fastapi "uvicorn[standard]" pydantic httpx

python

# main.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="My API", version="0.1.0")

class Item(BaseModel):
    name: str
    price: float

@app.get("/")
async def root():
    return {"hello": "world"}

@app.post("/items", response_model=Item, status_code=201)
async def create_item(item: Item) -> Item:
    return item

powershell

uv run uvicorn main:app --reload

Visit:

http://127.0.0.1:8000/docs — Swagger UI
http://127.0.0.1:8000/redoc — ReDoc
http://127.0.0.1:8000/openapi.json — OpenAPI schema

That's it. Typed Pydantic models become request validators, response serializers, and OpenAPI schema — automatically.

2. Path / query / body parameters

python

from fastapi import FastAPI, Path, Query, Body
from pydantic import BaseModel

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., ge=1, description="Item ID"),
    q: str | None = Query(None, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, gt=0, le=100),
):
    return {"item_id": item_id, "q": q, "skip": skip, "limit": limit}

class Item(BaseModel):
    name: str
    price: float

@app.post("/items")
async def create(item: Item, urgent: bool = False):
    return {"item": item, "urgent": urgent}

Path params come from the URL.
Query params are non-Pydantic args with default values.
Body comes from one (or more) Pydantic models.
Headers / cookies / forms via Header, Cookie, Form.

3. Response models, status codes, exceptions

python

from fastapi import HTTPException, status
from pydantic import BaseModel

class UserOut(BaseModel):
    id: int
    name: str
    # model_config = {"from_attributes": True}    # if mapping from ORM

@app.get("/users/{uid}", response_model=UserOut, status_code=200)
async def get_user(uid: int):
    user = await db.find(uid)
    if user is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="not found")
    return user

response_model ensures the response is filtered through that schema — extra fields are stripped, validation runs.

Custom exception handler

python

from fastapi import Request
from fastapi.responses import JSONResponse

class MyError(Exception):
    def __init__(self, code: str, msg: str):
        self.code = code; self.msg = msg

@app.exception_handler(MyError)
async def handle_my_error(request: Request, exc: MyError):
    return JSONResponse(status_code=400, content={"code": exc.code, "msg": exc.msg})

For uniform error envelopes, use a single handler + a custom exception family.

4. Dependencies — `Depends`

The cornerstone of FastAPI architecture. A dependency is a function whose return value is injected into endpoints.

python

from fastapi import Depends

async def get_db() -> AsyncSession:
    async with AsyncSession(engine) as s:
        yield s

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

Depends(get_db) runs get_db for each request, passes the result. The yield form is a setup/teardown context (closes the session after the response).

Reusable sub-dependencies

python

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    user = await db.get(User, payload["sub"])
    if not user: raise HTTPException(401)
    return user

@app.get("/me", response_model=UserOut)
async def me(user: User = Depends(get_current_user)):
    return user

get_current_user depends on oauth2_scheme and get_db. FastAPI builds the dependency graph; each dependency runs once per request even if multiple endpoints share it.

Class as dependency

python

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10): ...

@app.get("/items")
async def list_items(p: Pagination = Depends()):
    return ...

Concise; co-locates parameters with their validation.

Path-operation decorators (auth at route level)

python

@app.get("/admin", dependencies=[Depends(require_admin)])
async def admin_only():
    ...

When the dependency only has side effects (e.g., checking permissions) and the endpoint doesn't need the value.

5. Lifespan — startup / shutdown

python

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    app.state.engine = create_async_engine(DATABASE_URL)
    app.state.client = httpx.AsyncClient()
    yield
    # shutdown
    await app.state.client.aclose()
    await app.state.engine.dispose()

app = FastAPI(lifespan=lifespan)

Use lifespan for any resource that should be created once per process and torn down on shutdown — DB engines, HTTP clients, model loads, message-bus connections.

Access via dependency:

python

async def get_client(request: Request) -> httpx.AsyncClient:
    return request.app.state.client

6. Middleware

python

from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(GZipMiddleware, minimum_size=1000)

Custom middleware:

python

from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.3f}"
        return response

app.add_middleware(TimingMiddleware)

Common middleware: logging, request ID, tracing (OpenTelemetry), rate limiting, error reporting (Sentry).

7. Background tasks

python

from fastapi import BackgroundTasks

def send_email(to: str, body: str):
    smtp.send(to, body)

@app.post("/signup")
async def signup(user: SignupReq, tasks: BackgroundTasks):
    new = await create_user(user)
    tasks.add_task(send_email, user.email, "Welcome!")
    return new

Tasks run after the response is sent. Within the same process — fine for "tiny non-blocking work." For real jobs use Celery, arq, dramatiq, Redis Streams + worker, or a Cloud Tasks queue.

8. Streaming responses

python

from fastapi.responses import StreamingResponse

async def gen():
    for i in range(10):
        yield f"data: chunk-{i}\n\n"
        await asyncio.sleep(0.5)

@app.get("/stream")
async def stream():
    return StreamingResponse(gen(), media_type="text/event-stream")

text/event-stream is Server-Sent Events (SSE). For LLM token streaming, this is the standard pattern. Phase 8.4 covers it in depth.

9. Auth patterns

API key in header

python

from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(key: str = Depends(api_key_header)):
    if not is_valid(key): raise HTTPException(401)
    return key

@app.get("/secure", dependencies=[Depends(verify_api_key)])
async def secure(): ...

OAuth2 password flow + JWT

python

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt, time

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

@app.post("/token")
async def token(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form.username, form.password)
    if not user: raise HTTPException(401)
    token = jwt.encode(
        {"sub": user.id, "exp": time.time() + 3600},
        SECRET, algorithm="HS256",
    )
    return {"access_token": token, "token_type": "bearer"}

async def current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    return await db.get(User, payload["sub"])

For production: use authlib or fastapi-users for full flows, refresh tokens, social login.

Sessions

For HTML apps, store sessions in cookies via starlette.middleware.sessions or Redis-backed sessions.

10. Testing FastAPI

python

import pytest
import httpx
from main import app

@pytest.mark.asyncio
async def test_root():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/")
        assert r.status_code == 200
        assert r.json() == {"hello": "world"}

@pytest.mark.asyncio
async def test_create_item():
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
        r = await client.post("/items", json={"name": "x", "price": 1.0})
        assert r.status_code == 201
        assert r.json()["name"] == "x"

httpx.ASGITransport calls the app in-process (no network). Combine with app.dependency_overrides to inject fakes:

python

app.dependency_overrides[get_db] = lambda: InMemoryDB()

11. Logging + observability

python

import logging, structlog
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = structlog.get_logger()

FastAPIInstrumentor.instrument_app(app)         # adds OTel tracing

Add a request-ID middleware so every log line includes the request context:

python

import contextvars, uuid
request_id_ctx = contextvars.ContextVar("request_id", default="")

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        rid = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        token = request_id_ctx.set(rid)
        try:
            response = await call_next(request)
        finally:
            request_id_ctx.reset(token)
        response.headers["X-Request-ID"] = rid
        return response

Wire into structlog or your log formatter to print rid in every log line.

12. Production deployment

Run with multiple workers

powershell

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
# OR Gunicorn + uvicorn workers:
gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000

--workers N for CPU-bound workloads. Each worker is a separate process (multiplies memory but uses multiple cores).

Behind a reverse proxy

Caddy / nginx / Traefik in front handles TLS, gzip, rate limits.

Containerise

dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN pip install uv && uv sync --frozen --no-dev
COPY src ./src
CMD ["uv", "run", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Or use the official astral-sh/uv Docker image as a base. Phase 9.3 covers production Docker.

Async DB pool sizing

pool_size + max_overflow ≥ workers × concurrent_requests. Tune with load tests.

13. Other frameworks

Starlette: the ASGI toolkit FastAPI is built on. Use directly when you want minimal scaffolding.
Litestar (was Starlite): FastAPI-style with built-in OpenAPI, dependency injection, DTOs. Smaller community, sometimes faster.
Sanic, Quart: async Flask-like frameworks; less popular for new code.
Django + DRF: still dominant for full server-rendered web apps with ORM, auth, admin out of the box.
Flask: synchronous; great for tiny apps and microservices.

For new 2026 API code: FastAPI (or Litestar) for APIs; Django for full web apps.

14. Worked example: a tiny CRUD service

python

# src/app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic import BaseModel
import os

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./app.db")

class Base(DeclarativeBase): pass
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)

class UserIn(BaseModel): name: str; email: str
class UserOut(BaseModel):
    id: int; name: str; email: str
    model_config = {"from_attributes": True}

engine = create_async_engine(DATABASE_URL)

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

async def get_db():
    async with AsyncSession(engine) as s:
        yield s

@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(payload: UserIn, db: AsyncSession = Depends(get_db)):
    user = User(**payload.model_dump())
    db.add(user)
    try:
        await db.commit()
    except Exception:
        await db.rollback()
        raise HTTPException(409, "email exists")
    await db.refresh(user)
    return user

@app.get("/users/{uid}", response_model=UserOut)
async def get_user(uid: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, uid)
    if not user: raise HTTPException(404)
    return user

Run: uv run uvicorn src.app.main:app --reload. Visit /docs. Working CRUD in ~40 lines.

Hands-on lab (3 hours)

Build a CRUD endpoint set (POST/GET/PUT/DELETE) for one resource backed by SQLAlchemy async.
Add JWT auth with OAuth2PasswordBearer.
Add a lifespan that opens / closes an httpx.AsyncClient reused across requests.
Add a custom middleware that injects a request-id header and logs latency.
Add a StreamingResponse endpoint that yields 10 chunks.
Test endpoints with httpx.ASGITransport + pytest-asyncio.
Containerise with the Docker template; run; curl it.
Bonus: wire OpenTelemetry tracing and visualise in Jaeger / Datadog.

Common pitfalls

Blocking calls inside async def (e.g., requests.get, big numpy op) → freezes the loop.
Sharing one DB session across requests (race conditions) — use Depends(get_db) per-request.
Forgetting response_model= and leaking internal fields.
Big background tasks via BackgroundTasks instead of a real queue.
Loose CORS (allow_origins=["*"]) in production.
Storing secrets in code instead of env / secret manager.
Not handling DB constraint errors → 500s where 4xx belongs.

Self-check

What does Depends do?
When use lifespan vs middleware?
How would you stream chunks to a client?
How do you test endpoints in-process?
How would you scale a FastAPI app across cores?

References

FastAPI docs: https://fastapi.tiangolo.com/.
Starlette docs: https://www.starlette.io/.
Pydantic v2 docs: https://docs.pydantic.dev/latest/.
FastAPI: Modern Python Web Development, Bill Lubanovic.
Sebastián Ramírez (tiangolo) — talks and tutorials.

FastAPI in depth — routers, dependencies, and lifespan

Why this matters

Learning objectives

1. Hello FastAPI

2. Path / query / body parameters

3. Response models, status codes, exceptions

Custom exception handler

4. Dependencies — Depends

Reusable sub-dependencies

Class as dependency

Path-operation decorators (auth at route level)

5. Lifespan — startup / shutdown

6. Middleware

7. Background tasks

8. Streaming responses

9. Auth patterns

API key in header

OAuth2 password flow + JWT

Sessions

10. Testing FastAPI

11. Logging + observability

12. Production deployment

Run with multiple workers

Behind a reverse proxy

Containerise

Async DB pool sizing

13. Other frameworks

14. Worked example: a tiny CRUD service

Hands-on lab (3 hours)

Common pitfalls

Self-check

References

4. Dependencies — `Depends`