Staff Prep 17: FastAPI Internals — Routing, DI, and Performance Tuning

from fastapi import FastAPI, APIRouter, Depends

# api/routers/users.py
router = APIRouter(
    prefix="/users",
    tags=["users"],
    dependencies=[Depends(verify_token)],  # applied to all routes in this router
)

@router.get("/{user_id}")
async def get_user(user_id: int, db=Depends(get_db)):
    return await db.get(User, user_id)

@router.post("/", status_code=201)
async def create_user(user: UserCreate, db=Depends(get_db)):
    return await db.create(User, user.dict())

# api/routers/orders.py
orders_router = APIRouter(prefix="/orders", tags=["orders"])

# main.py
app = FastAPI()
app.include_router(users.router, prefix="/v1")
app.include_router(orders.orders_router, prefix="/v1")

# All routes: /v1/users/{id}, /v1/orders, etc.

from fastapi import Depends

# Default: cached per request — same instance reused
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# use_cache=False: new instance per injection point
# (rarely needed, but useful for testing isolation)
async def get_uncached_db(db: AsyncSession = Depends(get_db, use_cache=False)):
    return db

# Dependency with its own state: shared across requests
# Pattern: singleton dependency via module-level state
class DatabaseService:
    def __init__(self):
        self.pool = None

    async def startup(self):
        self.pool = await asyncpg.create_pool("postgresql://...")

db_service = DatabaseService()

@app.on_event("startup")
async def startup():
    await db_service.startup()

def get_db_service() -> DatabaseService:
    return db_service  # returns the same singleton

@app.get("/health")
async def health(db: DatabaseService = Depends(get_db_service)):
    await db.pool.fetchval("SELECT 1")
    return {"ok": True}

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request
import uuid
import time

# Use Middleware for:
# - Cross-cutting concerns (logging, timing, CORS, auth headers)
# - Things that apply to ALL routes including static files
# - Request/response transformation at the bytes level

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

# Use Dependencies for:
# - Business logic setup (DB sessions, current user, permissions)
# - Things that should only run for specific routes or routers
# - Things that need to be testable and mockable in isolation

async def get_current_user(token: str = Depends(get_token), db=Depends(get_db)):
    # Only runs for routes that declare this dependency
    # Can be easily mocked in tests
    return await verify_and_fetch_user(token, db)

# Key difference: middleware cannot access route handler return values easily
# Dependencies can see the full request context but not raw bytes

from pydantic import BaseModel
from typing import Optional

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    # NOTE: password, internal_flags NOT in response model
    # FastAPI filters these out automatically

@app.get("/users/{id}", response_model=UserResponse)
async def get_user(id: int, db=Depends(get_db)):
    user = await db.get(User, id)
    return user  # FastAPI serialises through UserResponse, strips excluded fields

# Performance: response_model validation adds overhead
# For hot paths returning large datasets:
@app.get("/users/bulk", response_model=None)  # skip validation
async def get_users_bulk(db=Depends(get_db)):
    users = await db.execute("SELECT id, name, email FROM users LIMIT 1000")
    from fastapi.responses import ORJSONResponse
    return ORJSONResponse(content=[dict(u) for u in users.fetchall()])
    # ORJSONResponse is 2-3x faster than default JSONResponse for large payloads

from fastapi.responses import StreamingResponse
import asyncio

# Stream large CSV export without loading everything into memory
async def generate_csv_stream(query_params: dict):
    yield "id,name,email
"  # header

    offset = 0
    batch_size = 1000

    while True:
        rows = await db.execute(
            "SELECT id, name, email FROM users LIMIT $1 OFFSET $2",
            batch_size, offset
        )
        batch = rows.fetchall()
        if not batch:
            break

        for row in batch:
            yield f"{row['id']},{row['name']},{row['email']}
"

        offset += batch_size
        await asyncio.sleep(0)  # yield to event loop

@app.get("/users/export.csv")
async def export_users():
    return StreamingResponse(
        generate_csv_stream({}),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"}
    )

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_failed",
            "fields": [
                {"path": " -> ".join(str(x) for x in err["loc"]),
                 "message": err["msg"]}
                for err in exc.errors()
            ]
        }
    )

class NotFoundError(Exception):
    def __init__(self, resource: str, id: int):
        self.resource = resource
        self.id = id

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
    return JSONResponse(
        status_code=404,
        content={"error": f"{exc.resource} with id {exc.id} not found"}
    )

@app.get("/orders/{id}")
async def get_order(id: int, db=Depends(get_db)):
    order = await db.get(Order, id)
    if not order:
        raise NotFoundError("Order", id)
    return order