BOJ/.claude/skills/code-review-skill/reference/fastapi.md

FastAPI Code Review Guide

FastAPI code review guide covering dependency injection (Depends), Pydantic v2 validation boundaries, async correctness, database session lifecycle and N+1, security, and a test-driven verification workflow that turns the reviewer's in-process test client into a tool for proving bugs rather than guessing at them.

Table of Contents

• Dependency Injection (Depends)
• Pydantic v2 Models & Validation
• Async Correctness
• Database Sessions & N+1
• Security
• Test-Driven Verification
• Review Checklist
• References

---

Dependency Injection (Depends)

FastAPI's Depends is the seam that keeps routes thin and testable. Most review problems here come from doing real work in the route function instead of behind a dependency.

Business logic belongs behind a dependency or service, not in the route

# ❌ Bad — DB access, auth, and business rules all inline in the route
@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    conn = await asyncpg.connect(DATABASE_URL)  # connection created per request
    row = await conn.fetchrow("SELECT * FROM orders WHERE id = $1", order_id)
    await conn.close()
    if row is None:
        raise HTTPException(404)
    return dict(row)

# ✅ Good — the route declares what it needs; the session is injected and pooled
async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session

@app.get("/orders/{order_id}", response_model=OrderOut)
async def get_order(order_id: int, session: AsyncSession = Depends(get_session)):
    order = await session.get(Order, order_id)
    if order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    return order

The injected version is also the version you can override in tests (see Test-Driven Verification).

yield dependencies must clean up, and cleanup runs even on error

# ❌ Bad — no cleanup; the session leaks if the route raises
async def get_session() -> AsyncSession:
    return SessionLocal()

# ✅ Good — the context manager closes the session on success AND on exception
async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session

Review point: confirm any yield dependency holding a resource (DB session, file handle, lock) releases it through a context manager or try/finally, so an exception in the route does not leak it.

Don't re-create singletons per request

# ❌ Bad — a new HTTP client (and connection pool) per request
@app.get("/proxy")
async def proxy(client: httpx.AsyncClient = Depends(lambda: httpx.AsyncClient())):
    ...

# ✅ Good — one client for the app lifetime, injected by reference
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient()
    yield
    await app.state.http.aclose()

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

Prefer the Annotated form and async dependencies

Since FastAPI 0.95 the idiomatic way to declare a dependency is Annotated[T, Depends(...)], not the default-value form. It is reusable across routes and plays well with type checkers. Also prefer async def dependencies: a sync (def) dependency runs in the threadpool, which is wasted overhead for a small non-I/O check.

# ⚠️ Older form — still works, but not the current idiom
@app.get("/items")
async def list_items(session: AsyncSession = Depends(get_session)): ...

# ✅ Good — Annotated form; define once, reuse everywhere
SessionDep = Annotated[AsyncSession, Depends(get_session)]

@app.get("/items")
async def list_items(session: SessionDep): ...

Use dependencies to validate existence and permissions — they're cached per request

A dependency is the natural place to answer "does this resource exist and may this caller touch it?" Pydantic validates shape; a dependency validates against the database. FastAPI caches each dependency's result within a single request, so chaining small dependencies costs nothing extra and removes duplicated lookups.

# ✅ Good — small dependencies chain; valid_post is resolved once per request
async def valid_post(post_id: int, session: SessionDep) -> Post:
    post = await session.get(Post, post_id)
    if post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

async def owned_post(post: Annotated[Post, Depends(valid_post)], user: CurrentUser) -> Post:
    if post.owner_id != user.id:
        raise HTTPException(status_code=403, detail="Forbidden")
    return post

@app.delete("/posts/{post_id}", status_code=204)
async def delete_post(post: Annotated[Post, Depends(owned_post)], session: SessionDep):
    await session.delete(post)        # existence + ownership already enforced
    await session.commit()

This is also the cleanest place to fix the auth-vs-authorization bug from the Security section: the ownership check moves into a reusable owned_post dependency.

---

Pydantic v2 Models & Validation

Separate input and output models; never echo the ORM object directly

# ❌ Bad — response_model is the DB model, so hashed_password leaks to the client
@app.post("/users", response_model=UserTable)
async def create_user(user: UserTable):  # also accepts client-set id, is_admin...
    ...

# ✅ Good — distinct schemas draw the trust boundary
class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserOut(BaseModel):
    id: int
    email: EmailStr
    model_config = ConfigDict(from_attributes=True)  # read from ORM safely

@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(payload: UserCreate, session: AsyncSession = Depends(get_session)):
    ...

response_model is a filter, not just documentation — fields absent from the output model are stripped from the response. Reusing the DB model as the response is the most common way sensitive fields leak.

Use distinct Create and Update schemas

# ❌ Bad — one schema for create and update means every field is required on PATCH
class ItemSchema(BaseModel):
    name: str
    price: float

# ✅ Good — update is a partial; create requires the full payload
class ItemCreate(BaseModel):
    name: str
    price: float = Field(gt=0)

class ItemUpdate(BaseModel):
    name: str | None = None
    price: float | None = Field(default=None, gt=0)

Validate at the boundary, not after the DB write

# ❌ Bad — negative quantity reaches the database before anything checks it
@app.post("/cart")
async def add_to_cart(item_id: int, quantity: int):
    await save(item_id, quantity)  # quantity = -5 silently accepted

# ✅ Good — the type system rejects it before the handler body runs
class CartLine(BaseModel):
    item_id: int
    quantity: int = Field(gt=0)

@app.post("/cart")
async def add_to_cart(line: CartLine):
    await save(line.item_id, line.quantity)

---

Async Correctness

This is the axis on which FastAPI differs most from Django and Flask, and the one most worth a reviewer's attention. FastAPI's throughput comes from a single event loop interleaving many concurrent requests. That model only holds if the loop is never blocked: one synchronous call on the loop stalls every in-flight request, not just its own. Get this wrong across the codebase and FastAPI does not just lose its edge — it performs worse than a sync framework like Flask, because Flask's worker-per-request model has no shared loop to choke. The reviewer's job is to keep work on the loop genuinely non-blocking and to treat every escape hatch as a cost, not a fix.

Never call blocking code inside an async def route

# ❌ Bad — blocking I/O on the loop freezes ALL concurrent requests, not just this one
@app.get("/report")
async def report():
    data = requests.get("https://slow-api.example.com").json()  # blocking socket
    time.sleep(2)                                                # blocks the loop
    return data

# ✅ Good — await a native-async client; the loop serves other requests meanwhile
@app.get("/report")
async def report(client: httpx.AsyncClient = Depends(get_http)):
    resp = await client.get("https://slow-api.example.com")
    return resp.json()

Prefer native-async SDKs over sync libraries

The right fix for blocking I/O is almost always a library that speaks async natively — not wrapping a sync one. Reach for the async client first; the threadpool is the last resort, not the default.

Sync (blocks the loop)	Native-async replacement
requests	httpx.AsyncClient, aiohttp
psycopg2 (sync)	asyncpg, SQLAlchemy async engine
redis-py (sync)	redis.asyncio
pymongo	motor
boto3	aioboto3

If you find asyncio.run(...), a new event loop, or a manually started thread inside a route, that is a red flag — it's an attempt to bolt sync code onto the loop. asyncio.run() inside a running loop raises RuntimeError outright; the rest quietly burns the performance you adopted FastAPI for.

# ❌ Bad — spinning up a loop/thread to call an async SDK from a sync context
@app.get("/users/{uid}")
def get_user(uid: int):
    return asyncio.run(repo.fetch(uid))  # RuntimeError under the running loop

# ✅ Good — let the route be async and await the native client directly
@app.get("/users/{uid}")
async def get_user(uid: int):
    return await repo.fetch(uid)

The threadpool is a bounded escape hatch, not a default

A plain def route — and run_in_threadpool(...) — does not run on the loop; FastAPI runs it in a bounded worker threadpool (AnyIO's default cap is 40 threads). For an occasional, genuinely-unavoidable blocking call this is the correct tool:

from fastapi.concurrency import run_in_threadpool

@app.get("/legacy")
async def legacy():
    return await run_in_threadpool(blocking_library_call)  # only if no async SDK exists

But it does not scale the way the loop does. Route every hot path through the threadpool and, under load, all workers block at once; further requests queue behind the cap and throughput collapses. Spawning your own threads or processes to "add concurrency" makes it worse: once live threads exceed the machine's core count, context-switch and GIL contention degrade performance sharply rather than improving it. The escape hatch is for the rare blocking dependency you cannot replace — not a substitute for choosing async SDKs.

Review heuristic: a def route is acceptable for a low-traffic endpoint with no async equivalent. A high-traffic endpoint doing blocking work in a def route (or via run_in_threadpool) is a scaling bug — flag it and ask for an async SDK.

CPU-bound work belongs in a worker process, not the loop or the threadpool

Neither the event loop nor the threadpool helps CPU-bound work: under the GIL only one thread runs Python bytecode at a time, so a heavy computation blocks just as badly from a threadpool as from the loop. Offload it to a separate process (Celery, Arq, RQ, or multiprocessing).

# ❌ Bad — a CPU-heavy job pins a worker; throughput drops for everyone
@app.post("/render")
async def render(doc: Doc):
    return heavy_pdf_render(doc)            # seconds of pure CPU on the loop

# ✅ Good — enqueue to a worker process; return a job handle
@app.post("/render", status_code=202)
async def render(doc: Doc):
    job = await queue.enqueue(heavy_pdf_render, doc)
    return {"job_id": job.id}

Don't fire-and-forget unawaited coroutines

# ❌ Bad — coroutine never awaited; the email is never sent (and no error surfaces)
@app.post("/signup")
async def signup(user: UserCreate):
    send_welcome_email(user.email)  # returns a coroutine, silently dropped

# ✅ Good — defer post-response work with BackgroundTasks
@app.post("/signup")
async def signup(user: UserCreate, tasks: BackgroundTasks):
    tasks.add_task(send_welcome_email, user.email)

BackgroundTasks runs in-process and offers no retries or persistence — use it only for short, fire-and-forget work (send an email, log an event). Anything long-running or retry-critical (data processing, payments) belongs in a real task queue (Celery/Arq/RQ).

---

Database Sessions & N+1

One session per request, injected — not a global

# ❌ Bad — a module-level session is shared across concurrent requests (not safe)
session = SessionLocal()

# ✅ Good — request-scoped session via dependency (see get_session above)
@app.get("/items")
async def list_items(session: AsyncSession = Depends(get_session)):
    ...

Eager-load relationships to avoid N+1

# ❌ Bad — one query for orders, then one query per order for its customer
orders = (await session.execute(select(Order))).scalars().all()
return [{"id": o.id, "customer": o.customer.name} for o in orders]  # N+1

# ✅ Good — a single query with the relationship eager-loaded
stmt = select(Order).options(selectinload(Order.customer))
orders = (await session.execute(stmt)).scalars().all()
return [{"id": o.id, "customer": o.customer.name} for o in orders]

With async SQLAlchemy, lazy attribute access outside the session often raises instead of silently querying — but the design issue is the same. Look for relationship access inside a loop without an options(...) eager load.

Paginate list endpoints

# ❌ Bad — returns every row; degrades as the table grows
@app.get("/users")
async def list_users(session: AsyncSession = Depends(get_session)):
    return (await session.execute(select(User))).scalars().all()

# ✅ Good — bounded page with a sane cap
@app.get("/users", response_model=list[UserOut])
async def list_users(
    session: AsyncSession = Depends(get_session),
    limit: int = Query(default=50, le=100),
    offset: int = Query(default=0, ge=0),
):
    stmt = select(User).limit(limit).offset(offset)
    return (await session.execute(stmt)).scalars().all()

Aggregate and join in SQL, not in Python

If a handler pulls rows into memory and then loops to group, count, or join them, the database is being used as dumb storage. Push the work down — the database does set operations far faster, and you transfer less data.

# ❌ Bad — fetch every order, then tally per customer in Python
orders = (await session.execute(select(Order))).scalars().all()
totals: dict[int, float] = {}
for o in orders:
    totals[o.customer_id] = totals.get(o.customer_id, 0) + o.amount

# ✅ Good — let the database group and sum
stmt = select(Order.customer_id, func.sum(Order.amount)).group_by(Order.customer_id)
totals = dict((await session.execute(stmt)).all())

---

Security

A declared auth dependency is not an enforced authorization check

This is the highest-value thing to look for. Depends(get_current_user) proves who the caller is — it does not prove they may touch this resource.

# ❌ Bad — any authenticated user can delete any other user's document
@app.delete("/documents/{doc_id}")
async def delete_document(
    doc_id: int,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    doc = await session.get(Document, doc_id)
    await session.delete(doc)            # never checks doc.owner_id == user.id
    await session.commit()

# ✅ Good — ownership is verified before the mutation
@app.delete("/documents/{doc_id}", status_code=204)
async def delete_document(
    doc_id: int,
    user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
):
    doc = await session.get(Document, doc_id)
    if doc is None:
        raise HTTPException(status_code=404, detail="Not found")
    if doc.owner_id != user.id:
        raise HTTPException(status_code=403, detail="Forbidden")
    await session.delete(doc)
    await session.commit()

The Test-Driven Verification section reproduces exactly this bug with a failing test.

Parameterize SQL; never f-string user input

# ❌ Bad — SQL injection
await session.execute(text(f"SELECT * FROM users WHERE email = '{email}'"))

# ✅ Good — bound parameter
await session.execute(text("SELECT * FROM users WHERE email = :email"), {"email": email})

Don't widen CORS to credentials + wildcard

# ❌ Bad — wildcard origin together with credentials is rejected by browsers and unsafe
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True)

# ✅ Good — enumerate trusted origins when credentials are allowed
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
)

Also check: secrets read from config/env (not hard-coded), HTTPException details that don't leak internals (stack traces, SQL), and rate limiting on auth endpoints.

---

Test-Driven Verification

Inspired by the test-driven development discipline: if you didn't watch the test fail, you don't know it tests the right thing. This matters even more for a coding agent than for a human reviewer. An agent's reading and reasoning are fallible — it can misread control flow, hallucinate a guarantee that isn't there, or rationalize a comfortable conclusion — so a prose verdict like "this looks safe" carries little weight on its own. An executable test is the one piece of objective ground truth the agent fully controls: it either passes or it doesn't, regardless of how confident the reasoning felt. That is what makes tests the agent's anchor of confidence. Reviewing the same way the discipline writes code — reproduce, don't assert — turns a hunch into proof.

A natural-language review comment ("this might let users delete each other's data") is exactly that kind of fallible hypothesis. FastAPI makes the ground truth cheap to obtain: an in-process client (httpx.AsyncClient over ASGITransport) runs the whole app, and app.dependency_overrides swaps out auth and the database without patching internals. So instead of trusting its own read of the code, the agent settles the question by reproduction.

Reproduce a suspected bug with a failing test (Verify RED)

Suppose the reviewer suspects the DELETE /documents/{doc_id} route above never checks ownership. Write the test that asserts the secure behavior, then run it and watch it fail — the failure is the proof.

# test_document_authorization.py
import pytest
from httpx import AsyncClient, ASGITransport
from fastapi import Header
from app.main import app
from app.deps import get_current_user, get_session

# Two users; the override picks one based on a test header.
USERS = {"alice": User(id=1, email="alice@example.com"),
         "bob":   User(id=2, email="bob@example.com")}

def fake_current_user(x_test_user: str = Header(default="alice")) -> User:
    return USERS[x_test_user]

@pytest.mark.asyncio
async def test_user_cannot_delete_another_users_document(session):  # async fixture
    # Arrange: a document owned by Alice (id=1)
    session.add(Document(id=10, owner_id=1, title="Alice's doc"))
    await session.commit()

    app.dependency_overrides[get_current_user] = fake_current_user
    app.dependency_overrides[get_session] = lambda: session

    # Act: Bob tries to delete Alice's document
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        resp = await client.delete("/documents/10", headers={"X-Test-User": "bob"})

    # Assert the SECURE behavior we expect
    assert resp.status_code == 403

    app.dependency_overrides.clear()

Run it against the unfixed code and confirm the failure is the bug, not a typo:

$ pytest test_document_authorization.py
FAILED  assert 204 == 403
#       ^ the endpoint deleted Alice's document for Bob — vulnerability confirmed

A failure of 204 == 403 (not an import error, not a 404) is what makes the finding credible: the route returned success for an action that should have been forbidden. Now the fix from the Security section turns it green:

$ pytest test_document_authorization.py
PASSED

Attach this test to the review. It documents the vulnerability, proves the fix, and guards against regression — far stronger than "consider checking ownership here."

Prefer dependency_overrides over patch/mock

FastAPI's DI is the seam the TDD discipline asks for: when something is hard to test without mocking everything, that usually signals coupling — and Depends already gives you the injection point, so you rarely need unittest.mock.patch.

# ❌ Bad — patching internals: brittle, couples the test to import paths
@patch("app.routes.orders.asyncpg.connect")
def test_get_order(mock_connect): ...

# ✅ Good — override the dependency with a real in-memory fake
app.dependency_overrides[get_session] = lambda: in_memory_session
app.dependency_overrides[get_current_user] = lambda: test_user

Always reset overrides between tests (app.dependency_overrides.clear() in a fixture teardown) so state doesn't leak across tests.

The reproduction above uses httpx.AsyncClient over ASGITransport with @pytest.mark.asyncio — the community convention for an async app, so the suite shares the app's event loop and you avoid loop-mismatch errors later. The synchronous TestClient is simpler and fine for a fully sync app, but standardizing on the async client from the start saves a painful migration once any route or fixture becomes async.

Critique the PR's own tests, not just its source

A PR that ships tests is not automatically safe. Apply these checks to the tests in the diff:

# ❌ Bad — happy-path only. Proves the route works when everything is correct,
#          says nothing about the validation and authorization paths.
def test_create_item():
    resp = client.post("/items", json={"name": "x", "price": 5})
    assert resp.status_code == 201

# ✅ Good — the boundary and failure paths are where bugs live
def test_create_item_rejects_negative_price():
    resp = client.post("/items", json={"name": "x", "price": -5})
    assert resp.status_code == 422

def test_create_item_requires_authentication():
    resp = client_without_auth.post("/items", json={"name": "x", "price": 5})
    assert resp.status_code == 401

Review questions for the test suite:

• Does it test behavior, or the mock? An assertion that only confirms a mock was called proves the test's own setup, not the endpoint.
• Are the failure paths covered? 401/403/404/422 — not just 200/201. Bugs cluster at the boundaries.
• Is the mock complete? A partial mock of an external API response that omits fields the handler reads passes in the test and fails in production.
• Were the tests written after the fact? Tests added alongside an implementation and passing on the first run never demonstrated that they can fail — and so prove little. A test that reproduces the bug (fails first, then passes) is worth more than one that was green from birth.

---

Review Checklist

Dependency Injection

• Routes stay thin — DB access and business rules live behind Depends/services
• yield dependencies release resources via context manager or try/finally
• Singletons (HTTP clients, pools) created once in lifespan, not per request
• Annotated[T, Depends(...)] form used; dependencies are async def unless they do blocking I/O
• Existence/permission checks live in (cached) dependencies, not copy-pasted into routes
• Dependencies are overridable in tests (no resources created inline in the route)

Validation

• Input and output use distinct Pydantic models; ORM objects are not the response_model
• response_model set so sensitive fields can't leak
• Separate Create vs Update schemas (update is partial)
• Constraints (gt, le, EmailStr, ...) enforced at the boundary, before the DB write

Async

• No blocking calls (requests, time.sleep, blocking DB drivers) inside async def
• Native-async SDKs preferred (httpx, asyncpg, redis.asyncio, ...) over sync ones
• No asyncio.run/manual event loops/manual threads inside routes
• run_in_threadpool/def routes used only as a last resort, not on hot paths
• CPU-bound work offloaded to a worker process (Celery/Arq/RQ), not the loop or threadpool
• No unawaited coroutines; BackgroundTasks only for short fire-and-forget work

Database

• One request-scoped session via dependency; no module-level shared session
• Relationships eager-loaded (selectinload/joinedload) where accessed in a loop
• Joins/aggregations done in SQL, not by looping in Python
• List endpoints are paginated with a capped limit

Security

• Authentication dependency is backed by an explicit authorization check (ownership/role)
• All SQL parameterized; no f-string interpolation of user input
• CORS does not combine allow_origins=["*"] with allow_credentials=True
• Secrets come from config/env; error responses don't leak internals

Tests

• Suspected bugs reproduced with a failing test (TestClient/AsyncClient) before being claimed
• dependency_overrides used instead of patching internals; overrides reset between tests
• Failure paths covered (401/403/404/422), not just the happy path
• Mocks of external responses are complete, not partial
• New tests demonstrate they can fail (reproduce-then-fix), not green from birth

---

References

• FastAPI official documentation — async, dependencies, testing
• zhanymkanov/fastapi-best-practices — production conventions (async routes, dependency caching, project structure)