long/BOJ
2
1
Fork 1
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity
Files
55cafdcd1d8c3756dcfc50df5bd73d28c53c560f
BOJ/.claude/skills/code-review-skill/reference/fastapi.md
T

585 lines
26 KiB
Markdown
Raw Blame History

# 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`)](#dependency-injection-depends)
- [Pydantic v2 Models & Validation](#pydantic-v2-models--validation)
- [Async Correctness](#async-correctness)
- [Database Sessions & N+1](#database-sessions--n1)
- [Security](#security)
- [Test-Driven Verification](#test-driven-verification)
- [Review Checklist](#review-checklist)
- [References](#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
```python
# ❌ 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](#test-driven-verification)).
### `yield` dependencies must clean up, and cleanup runs even on error
```python
# ❌ 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
```python
# ❌ 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.
```python
# ⚠️ 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.
```python
# ✅ 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](#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
```python
# ❌ 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
```python
# ❌ 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
```python
# ❌ 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
```python
# ❌ 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.
```python
# ❌ 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:
```python
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`).
```python
# ❌ 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
```python
# ❌ 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
```python
# ❌ 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
```python
# ❌ 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
```python
# ❌ 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.
```python
# ❌ 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.
```python
# ❌ 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](#test-driven-verification) section reproduces exactly this bug with a failing test.
### Parameterize SQL; never f-string user input
```python
# ❌ 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
```python
# ❌ 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.
```python
# 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:
```bash
$ 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](#security) section turns it green:
```bash
$ 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`.
```python
# ❌ 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:
```python
# ❌ 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](https://fastapi.tiangolo.com/) — async, dependencies, testing
- [zhanymkanov/fastapi-best-practices](https://github.com/zhanymkanov/fastapi-best-practices) — production conventions (async routes, dependency caching, project structure)
Reference in New Issue View Git Blame Copy Permalink