Fastapi Code Review

FastAPI Code Review

Quick Reference

Review Checklist

• [ ] APIRouter with proper prefix and tags
• [ ] All routes specify response_model for type safety
• [ ] Correct HTTP methods (GET, POST, PUT, DELETE, PATCH)
• [ ] Proper status codes (200, 201, 204, 404, etc.)
• [ ] Dependencies use Depends() not manual calls
• [ ] Yield dependencies have proper cleanup
• [ ] Request/Response models use Pydantic
• [ ] HTTPException with status code and detail
• [ ] All route handlers are async def
• [ ] No blocking I/O (requests, time.sleep, open())
• [ ] Background tasks for non-blocking operations
• [ ] No bare except in route handlers

Valid Patterns (Do NOT Flag)

These are idiomatic FastAPI patterns that may appear problematic but are correct:

• Pydantic validates request body automatically - No manual validation needed when using typed Pydantic models as parameters
• Dependency injection for database sessions - Sessions come from Depends(), not passed as function arguments
• HTTPException for all HTTP errors - FastAPI handles conversion to proper HTTP responses
• Async def endpoint without await - May be using sync dependencies or simple operations; FastAPI handles this
• Type annotation on Depends() - This is documentation/IDE support, not a type assertion
• Query/Path/Body defaults - FastAPI processes these at runtime, not traditional Python defaults
• Returning dict from endpoint - Pydantic converts automatically if response_model is set

Context-Sensitive Rules

Only flag issues when the context warrants it:

• Flag missing validation ONLY IF the field isn't already in a Pydantic model with validators
• Flag missing auth ONLY IF the endpoint isn't using Depends() with an auth dependency
• Flag missing error handling ONLY IF HTTPException isn't raised appropriately for error cases
• Flag sync in async ONLY IF the operation is actually blocking (file I/O, network calls, CPU-bound), not just non-async

Gates (FastAPI-specific)

Run once per FastAPI-related finding, after you can anchor `file:line` for the handler (see review-verification-protocol) and before the finding text ships. If a step’s pass condition is not met, do not assert the finding as written—gather evidence, withdraw, downgrade severity, or rephrase as a question.

Gate 1 — Route decorator and response surface

Gate 2 — Blocking or “should be async”

Gate 3 — Depends, validation, auth

FastAPI Framework Behaviors

FastAPI + Pydantic handle many concerns automatically:

• Request validation via Pydantic models
• Response serialization via response_model
• Dependency injection for cross-cutting concerns
• Exception handling via exception handlers

Before flagging "missing" functionality, verify FastAPI isn't handling it.

When to Load References

• Reviewing route definitions → routes.md
• Reviewing dependency injection → dependencies.md
• Reviewing Pydantic models/validation → validation.md
• Reviewing async route handlers → async.md

Review Questions

1. Do all routes have explicit response models and status codes? 2. Are dependencies injected via Depends() with proper cleanup? 3. Do all Pydantic models validate inputs correctly? 4. Are all route handlers async and non-blocking?

Before Submitting Findings

1. For each FastAPI-related finding, complete Gates (FastAPI-specific) above. 2. Load and follow review-verification-protocol (Pre-Report checklist and Verification by Issue Type) before reporting any issue.

Async

Critical Anti-Patterns

1. Blocking I/O in Async Handlers

Problem: Blocks the event loop, prevents concurrent request handling.

# BAD - blocking HTTP client
import requests

@router.get("/external")
async def fetch_external():
    response = requests.get("https://api.example.com")  # BLOCKS!
    return response.json()

# GOOD - async HTTP client
import httpx

@router.get("/external")
async def fetch_external():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com")
    return response.json()

2. Blocking Database Calls

Problem: Synchronous DB driver blocks event loop.

# BAD - sync SQLAlchemy
from sqlalchemy.orm import Session

@router.get("/users", response_model=list[UserResponse])
async def list_users(db: Session = Depends(get_db)):
    users = db.query(User).all()  # BLOCKS!
    return users

# GOOD - async SQLAlchemy
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

@router.get("/users", response_model=list[UserResponse])
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return users

3. Using time.sleep Instead of asyncio.sleep

Problem: Blocks event loop during sleep.

# BAD - blocking sleep
import time

@router.post("/jobs")
async def create_job():
    time.sleep(5)  # BLOCKS for 5 seconds!
    return {"status": "done"}

# GOOD - async sleep
import asyncio

@router.post("/jobs")
async def create_job():
    await asyncio.sleep(5)  # Yields control
    return {"status": "done"}

# BETTER - use background tasks for long operations
from fastapi import BackgroundTasks

async def process_job():
    await asyncio.sleep(5)
    # Do actual work

@router.post("/jobs")
async def create_job(background_tasks: BackgroundTasks):
    background_tasks.add_task(process_job)
    return {"status": "processing"}

4. Sync File I/O in Async Handlers

Problem: File operations block event loop.

# BAD - blocking file I/O
@router.get("/config")
async def get_config():
    with open("config.json") as f:  # BLOCKS!
        return json.load(f)

# GOOD - async file I/O
import aiofiles

@router.get("/config")
async def get_config():
    async with aiofiles.open("config.json") as f:
        content = await f.read()
    return json.loads(content)

# ACCEPTABLE - small files in executor
import asyncio

def read_config_sync():
    with open("config.json") as f:
        return json.load(f)

@router.get("/config")
async def get_config():
    loop = asyncio.get_event_loop()
    config = await loop.run_in_executor(None, read_config_sync)
    return config

5. Not Using Background Tasks

Problem: Long operations block response, timeout issues.

# BAD - blocks response
@router.post("/emails")
async def send_email(email: EmailCreate):
    await send_email_via_smtp(email)  # Takes 5 seconds!
    await log_email_sent(email)  # Takes 1 second!
    return {"status": "sent"}

# GOOD - use background tasks
from fastapi import BackgroundTasks

async def send_email_background(email: EmailCreate):
    await send_email_via_smtp(email)
    await log_email_sent(email)

@router.post("/emails", status_code=202)
async def send_email(
    email: EmailCreate,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email_background, email)
    return {"status": "queued"}

6. Sequential Instead of Concurrent Calls

Problem: Misses parallelization opportunity.

# BAD - sequential (slow)
@router.get("/dashboard")
async def get_dashboard(user_id: int):
    user = await get_user(user_id)
    posts = await get_user_posts(user_id)
    stats = await get_user_stats(user_id)
    return {"user": user, "posts": posts, "stats": stats}

# GOOD - concurrent (fast)
import asyncio

@router.get("/dashboard")
async def get_dashboard(user_id: int):
    user, posts, stats = await asyncio.gather(
        get_user(user_id),
        get_user_posts(user_id),
        get_user_stats(user_id)
    )
    return {"user": user, "posts": posts, "stats": stats}

7. Mixing Sync and Async Route Handlers

Problem: Inconsistent patterns, sync handlers block thread pool.

# BAD - mixing sync and async
@router.get("/sync-route")
def sync_handler():  # Blocks thread pool
    return db.query(User).all()

@router.get("/async-route")
async def async_handler():
    return await db.query_async(User)

# GOOD - all async
@router.get("/route1")
async def handler1():
    result = await db.execute(select(User))
    return result.scalars().all()

@router.get("/route2")
async def handler2():
    result = await db.execute(select(Post))
    return result.scalars().all()

8. Not Awaiting Coroutines

Problem: Coroutine never executes, silent failures.

# BAD - missing await
@router.post("/users")
async def create_user(user: UserCreate):
    db.create_user(user)  # Returns coroutine, doesn't execute!
    return {"status": "created"}  # User not actually created!

# GOOD - await coroutines
@router.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    created_user = await db.create_user(user)
    return created_user

9. Blocking External API Calls

Problem: Synchronous requests library blocks event loop.

# BAD - requests blocks
import requests

@router.get("/weather")
async def get_weather(city: str):
    response = requests.get(f"https://api.weather.com/{city}")  # BLOCKS!
    return response.json()

# GOOD - httpx async
import httpx

@router.get("/weather")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.weather.com/{city}")
    return response.json()

# GOOD - with timeout
@router.get("/weather")
async def get_weather(city: str):
    async with httpx.AsyncClient(timeout=5.0) as client:
        try:
            response = await client.get(f"https://api.weather.com/{city}")
            return response.json()
        except httpx.TimeoutException:
            raise HTTPException(504, detail="Weather API timeout")

Review Questions

1. Are all route handlers async def? 2. Are there any requests, time.sleep, or open() calls? 3. Is the database driver async (AsyncSession, asyncpg, etc.)? 4. Are background tasks used for long operations? 5. Are independent async calls parallelized with gather()? 6. Are all coroutines properly awaited? 7. Are external API calls using async HTTP clients?

Dependencies

Critical Anti-Patterns

1. Manual Dependency Calls

Problem: Bypasses FastAPI's injection system, no automatic cleanup.

# BAD - manually calling dependency
async def get_db_session():
    session = SessionLocal()
    return session

@router.get("/users")
async def list_users():
    db = await get_db_session()  # Manual call!
    users = await db.query(User).all()
    return users

# GOOD - using Depends()
from fastapi import Depends

async def get_db_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        await session.close()

@router.get("/users", response_model=list[UserResponse])
async def list_users(db: Session = Depends(get_db_session)):
    users = await db.query(User).all()
    return users

2. Missing Cleanup in Yield Dependencies

Problem: Resources leak, connections not closed.

# BAD - no cleanup
async def get_db():
    db = DatabaseConnection()
    yield db
    # Connection never closed!

# GOOD - proper cleanup
async def get_db():
    db = DatabaseConnection()
    try:
        yield db
    finally:
        await db.close()

3. Shared State Without Proper Scope

Problem: Dependencies create shared mutable state across requests.

# BAD - shared mutable state
cache = {}  # Shared across all requests!

async def get_cache():
    return cache

@router.get("/items/{id}")
async def get_item(id: int, cache: dict = Depends(get_cache)):
    # Multiple requests share same dict - race conditions!
    if id not in cache:
        cache[id] = await fetch_item(id)
    return cache[id]

# GOOD - request-scoped state
from contextvars import ContextVar

request_cache: ContextVar[dict] = ContextVar('request_cache')

async def get_cache():
    cache = {}
    request_cache.set(cache)
    return cache

# BETTER - use proper caching library
from functools import lru_cache

@lru_cache(maxsize=128)
async def get_item_cached(id: int):
    return await fetch_item(id)

4. Nested Depends Not Utilized

Problem: Duplicate code, no composition of dependencies.

# BAD - duplicated logic
async def get_current_user(token: str):
    # Verify token, decode, fetch user
    return user

async def get_admin_user(token: str):
    # Same verification, then check admin
    user = await verify_and_decode(token)
    if not user.is_admin:
        raise HTTPException(403)
    return user

# GOOD - compose dependencies
async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = await verify_token(token)
    if not user:
        raise HTTPException(401, detail="Invalid token")
    return user

async def get_admin_user(user: User = Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(403, detail="Admin required")
    return user

5. Dependencies with Side Effects

Problem: Dependencies modify state instead of providing resources.

# BAD - dependency has side effects
async def log_request(request: Request):
    # Side effect: writes to database
    await db.log_request(request)
    return None

@router.get("/users")
async def list_users(_: None = Depends(log_request)):
    return users

# GOOD - use middleware for cross-cutting concerns
from starlette.middleware.base import BaseHTTPMiddleware

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        await db.log_request(request)
        response = await call_next(request)
        return response

app.add_middleware(LoggingMiddleware)

# OR - dependency returns resource
async def get_logger(request: Request):
    logger = RequestLogger(request)
    return logger

@router.get("/users")
async def list_users(logger: RequestLogger = Depends(get_logger)):
    logger.info("Listing users")
    return users

6. Class-Based Dependencies Without Caching

Problem: New instance created unnecessarily.

# BAD - new instance every time
class DatabaseService:
    def __init__(self):
        self.connection_pool = create_pool()  # Expensive!

@router.get("/users")
async def list_users(db: DatabaseService = Depends(DatabaseService)):
    return await db.query_users()

# GOOD - use singleton or app state
class DatabaseService:
    def __init__(self, pool):
        self.pool = pool

async def get_db_service(
    pool = Depends(lambda: app.state.db_pool)
) -> DatabaseService:
    return DatabaseService(pool)

# OR - use dependency with cache
async def get_db_service() -> DatabaseService:
    return app.state.db_service

@router.get("/users")
async def list_users(db: DatabaseService = Depends(get_db_service)):
    return await db.query_users()

7. Security Dependencies Not Applied Globally

Problem: Easy to forget security on new routes.

# BAD - must remember to add auth to every route
@router.get("/users", dependencies=[Depends(verify_token)])
async def list_users(): ...

@router.get("/posts")  # Forgot auth!
async def list_posts(): ...

# GOOD - apply at router level
router = APIRouter(
    prefix="/api/v1",
    dependencies=[Depends(verify_token)]
)

@router.get("/users")
async def list_users(): ...

@router.get("/posts")
async def list_posts(): ...

Review Questions

1. Are all dependencies injected via Depends() not manually called? 2. Do yield dependencies have proper try/finally cleanup? 3. Is there any shared mutable state across requests? 4. Are nested dependencies used to compose common patterns? 5. Do dependencies provide resources, not perform side effects? 6. Are security dependencies applied at router or app level?

Routes

Critical Anti-Patterns

1. Missing response_model

Problem: No type safety, documentation unclear, response not validated.

# BAD
@router.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": "Alice"}

# GOOD
@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    return {"id": user_id, "name": "Alice"}

2. No APIRouter Prefix/Tags

Problem: Routes not organized, duplicated path prefixes, unclear docs.

# BAD
@app.get("/api/v1/users")
async def list_users(): ...

@app.get("/api/v1/users/{id}")
async def get_user(id: int): ...

# GOOD
router = APIRouter(prefix="/api/v1/users", tags=["users"])

@router.get("")
async def list_users(): ...

@router.get("/{id}")
async def get_user(id: int): ...

app.include_router(router)

3. Wrong HTTP Methods

Problem: Violates REST conventions, confusing semantics.

# BAD - using GET for mutations
@router.get("/users/{id}/delete")
async def delete_user(id: int): ...

# BAD - using POST for retrieval
@router.post("/users/{id}")
async def get_user(id: int): ...

# GOOD
@router.delete("/users/{id}", status_code=204)
async def delete_user(id: int): ...

@router.get("/users/{id}", response_model=UserResponse)
async def get_user(id: int): ...

4. Missing Status Codes

Problem: Always returns 200, even for creates/deletes.

# BAD - creates should return 201
@router.post("/users")
async def create_user(user: UserCreate):
    return created_user

# BAD - deletes should return 204
@router.delete("/users/{id}")
async def delete_user(id: int):
    return {"message": "deleted"}

# GOOD
@router.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    return created_user

@router.delete("/users/{id}", status_code=204)
async def delete_user(id: int):
    # 204 returns no content
    return None

5. Direct Exception Raising

Problem: Returns generic 500 errors instead of proper HTTP status codes.

# BAD
@router.get("/users/{id}")
async def get_user(id: int):
    user = await db.get_user(id)
    if not user:
        raise ValueError("User not found")
    return user

# GOOD
from fastapi import HTTPException

@router.get("/users/{id}", response_model=UserResponse)
async def get_user(id: int):
    user = await db.get_user(id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

6. Multiple Response Models

Problem: Same endpoint returns different schemas.

# BAD
@router.get("/users/{id}")
async def get_user(id: int, full: bool = False):
    if full:
        return UserDetailResponse(...)
    return UserSummaryResponse(...)

# GOOD - use separate endpoints
@router.get("/users/{id}", response_model=UserSummaryResponse)
async def get_user(id: int):
    return UserSummaryResponse(...)

@router.get("/users/{id}/full", response_model=UserDetailResponse)
async def get_user_full(id: int):
    return UserDetailResponse(...)

# ALTERNATIVE - use response_model with Union
from typing import Union

@router.get("/users/{id}", response_model=Union[UserSummaryResponse, UserDetailResponse])
async def get_user(id: int, full: bool = False):
    if full:
        return UserDetailResponse(...)
    return UserSummaryResponse(...)

7. Path Parameter Validation

Problem: No validation on path parameters.

# BAD
@router.get("/users/{user_id}")
async def get_user(user_id: int):
    # What if user_id is negative or zero?
    return await db.get_user(user_id)

# GOOD
from fastapi import Path

@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int = Path(..., gt=0)):
    return await db.get_user(user_id)

Review Questions

1. Does every route have an explicit response_model? 2. Are routes organized with APIRouter using prefix and tags? 3. Are HTTP methods semantically correct (GET for read, POST for create, etc.)? 4. Do create operations return 201? Do deletes return 204? 5. Are HTTPExceptions used instead of generic exceptions? 6. Are path parameters validated?

Validation

Critical Anti-Patterns

1. Manual Validation Instead of Pydantic

Problem: Duplicate validation logic, inconsistent errors.

# BAD - manual validation
@router.post("/users")
async def create_user(request: Request):
    data = await request.json()
    if "email" not in data:
        raise HTTPException(400, "Email required")
    if "@" not in data["email"]:
        raise HTTPException(400, "Invalid email")
    return await db.create_user(data)

# GOOD - Pydantic validation
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    name: str
    age: int | None = None

@router.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    return await db.create_user(user)

2. Missing Field Validators

Problem: Invalid data passes through.

# BAD - no validation on age
class UserCreate(BaseModel):
    name: str
    age: int  # Can be negative!

# GOOD - field validation
from pydantic import BaseModel, Field

class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    age: int = Field(..., ge=0, le=150)
    email: EmailStr

3. Generic HTTPException Messages

Problem: Users don't know what's wrong.

# BAD - vague error
@router.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(404)  # No detail!
    return user

# GOOD - specific error
@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(
            status_code=404,
            detail=f"User {user_id} not found"
        )
    return user

4. Not Using Pydantic Config

Problem: Models accept extra fields, expose internal fields.

# BAD - accepts any extra fields
class UserCreate(BaseModel):
    name: str
    email: str
    # {"name": "Alice", "email": "a@b.com", "is_admin": true} accepted!

# GOOD - strict validation
class UserCreate(BaseModel):
    name: str
    email: EmailStr

    class Config:
        extra = "forbid"  # Reject unknown fields

# GOOD - control ORM exposure
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    # Don't expose password_hash, created_at, etc.

    class Config:
        from_attributes = True  # Formerly orm_mode

5. Missing Custom Validators

Problem: Business rules not enforced.

# BAD - no validation
class PasswordReset(BaseModel):
    password: str
    confirm_password: str
    # Passwords might not match!

# GOOD - custom validator
from pydantic import BaseModel, model_validator

class PasswordReset(BaseModel):
    password: str = Field(..., min_length=8)
    confirm_password: str

    @model_validator(mode='after')
    def passwords_match(self):
        if self.password != self.confirm_password:
            raise ValueError('Passwords do not match')
        return self

6. Not Handling 422 Validation Errors

Problem: Default 422 responses unclear to clients.

# BAD - default 422 response is verbose and unclear
# (No custom handler)

# GOOD - custom 422 handler
from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError
):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(x) for x in error["loc"][1:]),
            "message": error["msg"],
            "type": error["type"]
        })

    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": errors}
    )

7. Using Dict Instead of Models

Problem: No validation, no type safety, unclear API.

# BAD - dict responses
@router.get("/users/{id}")
async def get_user(id: int) -> dict:
    return {
        "id": id,
        "name": "Alice",
        "extra_field": "oops"  # Inconsistent!
    }

# GOOD - Pydantic response model
class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@router.get("/users/{id}", response_model=UserResponse)
async def get_user(id: int):
    user = await db.get_user(id)
    if not user:
        raise HTTPException(404, detail="User not found")
    return user  # Auto-validates and filters fields

8. Missing Query Parameter Validation

Problem: Invalid query parameters not validated.

# BAD - no validation
@router.get("/users")
async def list_users(page: int = 1, size: int = 10):
    # What if page is 0 or negative?
    # What if size is 10000?
    return await db.get_users(page, size)

# GOOD - validated query params
from fastapi import Query

@router.get("/users", response_model=list[UserResponse])
async def list_users(
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100)
):
    return await db.get_users(page, size)

Review Questions

1. Are all request bodies defined as Pydantic models? 2. Do fields have proper validators (min_length, ge, EmailStr, etc.)? 3. Do HTTPExceptions include detailed error messages? 4. Are models configured with extra = "forbid" to reject unknown fields? 5. Are custom validators used for business rules? 6. Are query parameters validated with Query()? 7. Are response models used instead of plain dicts?