# Best Practices > Python FastAPI performance optimization and best practices guidelines Use when this capability is needed. - Skill: `tomevault-io/best-practices-17` (Agent Skill, multi-file: 2 files) - Install (CLI): `npx skillmds@latest add tomevault-io/best-practices-17` - Raw SKILL.md: https://api.skillmd.com/api/skills/tomevault-io/best-practices-17/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - Author: tomevault-io (https://skillmd.com/u/tomevault-io) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/tomevault-io/best-practices-17 --- # Python FastAPI Best Practices Comprehensive performance optimization and best practices for FastAPI applications. **48 rules across 8 categories** based on FastAPI official documentation (benchmark score 96.8), Pydantic V2, and SQLAlchemy async patterns. ## Activation Triggers This skill activates when: - Writing new endpoints, routers, or dependencies - Implementing async patterns and database operations - Optimizing API performance - Reviewing or refactoring FastAPI code - Debugging async, validation, or database issues - Implementing authentication/authorization - Writing tests for FastAPI applications ## Rule Categories | # | Category | Priority | Prefix | Rule Count | Focus | |---|----------|----------|--------|------------|-------| | 1 | Async Patterns | CRITICAL | `async-` | 6 | Event loop, blocking I/O | | 2 | Dependency Injection | CRITICAL | `di-` | 6 | Dependencies, lifecycle, scopes | | 3 | Pydantic Models | HIGH | `pydantic-` | 6 | Validation, serialization | | 4 | Database Patterns | HIGH | `db-` | 6 | SQLAlchemy async, sessions | | 5 | Error Handling | MEDIUM | `err-` | 6 | HTTPException, handlers | | 6 | Security | MEDIUM | `sec-` | 6 | Auth, CORS, rate limiting | | 7 | Performance | MEDIUM | `perf-` | 6 | Response, caching, background | | 8 | Testing | LOW | `test-` | 6 | pytest-asyncio, fixtures | ## Quick Reference - All 48 Rules ### Async Patterns (CRITICAL) - `async-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `async-block` | Never block the event loop | Prevents server freeze | | `async-gather` | Use asyncio.gather for parallel I/O | 2-10x speedup | | `async-timeout` | Always set async operation timeouts | Prevents hung requests | | `async-context` | Use async context managers | Proper resource cleanup | | `async-driver` | Use async database/HTTP drivers | Full async stack | | `async-executor` | Run blocking code in thread pool | CPU-bound isolation | ### Dependency Injection (CRITICAL) - `di-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `di-depends` | Use Depends() for shared logic | Code reuse, testability | | `di-yield` | Use yield for cleanup | Proper resource lifecycle | | `di-cache` | Cache expensive dependencies | Reduce redundant computation | | `di-scope` | Understand dependency scopes | Request vs app lifetime | | `di-override` | Use dependency_overrides for testing | Clean test isolation | | `di-annotated` | Use Annotated for cleaner signatures | Type-safe DI | ### Pydantic Models (HIGH) - `pydantic-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `pydantic-field` | Use Field() with constraints | Automatic validation | | `pydantic-validator` | Use @field_validator for custom logic | Complex validation | | `pydantic-config` | Configure model behavior | Performance tuning | | `pydantic-orm` | Use from_attributes for ORM | Clean serialization | | `pydantic-generic` | Use Generic models for reuse | DRY response models | | `pydantic-alias` | Use alias for API compatibility | Clean external APIs | ### Database Patterns (HIGH) - `db-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `db-async` | Use async SQLAlchemy engine | Non-blocking DB | | `db-session` | Use async session context manager | Connection safety | | `db-n-plus-one` | Avoid N+1 with eager loading | Query optimization | | `db-transaction` | Use explicit transactions | Data consistency | | `db-migration` | Use Alembic async migrations | Schema management | | `db-pool` | Configure connection pool properly | Resource efficiency | ### Error Handling (MEDIUM) - `err-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `err-http` | Use HTTPException with status codes | Proper HTTP semantics | | `err-handler` | Register global exception handlers | Centralized errors | | `err-validation` | Handle RequestValidationError | User-friendly messages | | `err-middleware` | Use middleware for cross-cutting | Consistent error format | | `err-logging` | Log errors with context | Debugging support | | `err-retry` | Implement retry with backoff | Resilience | ### Security (MEDIUM) - `sec-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `sec-oauth2` | Use OAuth2PasswordBearer | Standard auth flow | | `sec-jwt` | Implement JWT properly | Stateless auth | | `sec-cors` | Configure CORS correctly | Cross-origin security | | `sec-rate-limit` | Implement rate limiting | DoS protection | | `sec-input` | Validate and sanitize input | Injection prevention | | `sec-headers` | Set security headers | Browser protection | ### Performance (MEDIUM) - `perf-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `perf-response` | Use response_model for filtering | Reduce payload size | | `perf-cache` | Implement caching strategies | Reduce DB load | | `perf-background` | Use BackgroundTasks for async work | Faster responses | | `perf-streaming` | Use StreamingResponse for large data | Memory efficiency | | `perf-compression` | Enable response compression | Network efficiency | | `perf-middleware` | Order middleware by frequency | Reduce overhead | ### Testing (LOW) - `test-*` | Rule ID | Rule Name | Impact | |---------|-----------|--------| | `test-client` | Use TestClient or httpx.AsyncClient | Proper API testing | | `test-override` | Use dependency_overrides | Clean mocking | | `test-async` | Use pytest-asyncio properly | Async test support | | `test-fixture` | Create reusable fixtures | Test organization | | `test-mock` | Mock external services | Test isolation | | `test-coverage` | Maintain test coverage | Quality assurance | --- ## Detailed Rules by Category ### 1. Async Patterns (CRITICAL) #### `async-block` - Never Block the Event Loop **Impact**: Server freeze, request timeouts, degraded throughput ```python # ❌ INCORRECT - blocking I/O in async function import time import requests @app.get("/users/{user_id}") async def get_user(user_id: int): time.sleep(1) # BLOCKS entire server! response = requests.get(f"http://api/users/{user_id}") # BLOCKS! return response.json() # ✅ CORRECT - use async alternatives import asyncio import httpx @app.get("/users/{user_id}") async def get_user(user_id: int): await asyncio.sleep(1) # non-blocking async with httpx.AsyncClient() as client: response = await client.get(f"http://api/users/{user_id}") return response.json() ``` #### `async-gather` - Use asyncio.gather for Parallel I/O **Impact**: 2-10x speedup for independent operations ```python # ❌ INCORRECT - sequential fetches (3 round trips) @app.get("/dashboard") async def get_dashboard(user_id: int): user = await get_user(user_id) # 100ms orders = await get_orders(user_id) # 100ms notifications = await get_notifications(user_id) # 100ms return {"user": user, "orders": orders, "notifications": notifications} # Total: ~300ms # ✅ CORRECT - parallel fetches (1 round trip) @app.get("/dashboard") async def get_dashboard(user_id: int): user, orders, notifications = await asyncio.gather( get_user(user_id), get_orders(user_id), get_notifications(user_id) ) return {"user": user, "orders": orders, "notifications": notifications} # Total: ~100ms (max of all) ``` #### `async-timeout` - Always Set Async Operation Timeouts **Impact**: Prevents hung requests and resource exhaustion ```python # ❌ INCORRECT - no timeout @app.get("/external-api") async def call_external(): async with httpx.AsyncClient() as client: response = await client.get("http://slow-api.com/data") return response.json() # ✅ CORRECT - explicit timeouts from asyncio import timeout @app.get("/external-api") async def call_external(): async with httpx.AsyncClient(timeout=10.0) as client: async with timeout(15.0): # overall timeout response = await client.get("http://slow-api.com/data") return response.json() ``` #### `async-context` - Use Async Context Managers **Impact**: Proper resource cleanup, connection management ```python # ❌ INCORRECT - manual resource management @app.get("/data") async def get_data(): client = httpx.AsyncClient() response = await client.get("http://api/data") # client never closed! return response.json() # ✅ CORRECT - async context manager @app.get("/data") async def get_data(): async with httpx.AsyncClient() as client: response = await client.get("http://api/data") return response.json() # client properly closed ``` #### `async-driver` - Use Async Database/HTTP Drivers **Impact**: Full async stack, maximum concurrency ```python # ❌ INCORRECT - sync database driver blocks event loop from sqlalchemy import create_engine engine = create_engine("postgresql://user:pass@localhost/db") # ✅ CORRECT - async database driver from sqlalchemy.ext.asyncio import create_async_engine engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") # ❌ INCORRECT - sync HTTP client import requests response = requests.get(url) # ✅ CORRECT - async HTTP client import httpx async with httpx.AsyncClient() as client: response = await client.get(url) ``` #### `async-executor` - Run Blocking Code in Thread Pool **Impact**: Isolates CPU-bound work from async event loop ```python # ❌ INCORRECT - CPU-bound work blocks event loop import hashlib @app.post("/hash") async def compute_hash(data: bytes): result = hashlib.pbkdf2_hmac('sha256', data, b'salt', 100000) return {"hash": result.hex()} # ✅ CORRECT - run in thread pool executor import asyncio from concurrent.futures import ThreadPoolExecutor executor = ThreadPoolExecutor(max_workers=4) def blocking_hash(data: bytes) -> bytes: return hashlib.pbkdf2_hmac('sha256', data, b'salt', 100000) @app.post("/hash") async def compute_hash(data: bytes): loop = asyncio.get_event_loop() result = await loop.run_in_executor(executor, blocking_hash, data) return {"hash": result.hex()} ``` --- ### 2. Dependency Injection (CRITICAL) #### `di-depends` - Use Depends() for Shared Logic **Impact**: Code reuse, testability, separation of concerns ```python # ❌ INCORRECT - repeated code @app.get("/items/") async def get_items(skip: int = 0, limit: int = 10, db: AsyncSession = ...): # pagination logic repeated everywhere pass @app.get("/users/") async def get_users(skip: int = 0, limit: int = 10, db: AsyncSession = ...): # same pagination logic again pass # ✅ CORRECT - extracted dependency from fastapi import Depends, Query async def pagination( skip: int = Query(default=0, ge=0), limit: int = Query(default=10, ge=1, le=100) ) -> dict: return {"skip": skip, "limit": limit} @app.get("/items/") async def get_items(pagination: dict = Depends(pagination)): pass @app.get("/users/") async def get_users(pagination: dict = Depends(pagination)): pass ``` #### `di-yield` - Use yield for Cleanup **Impact**: Proper resource lifecycle, no leaks ```python # ❌ INCORRECT - no cleanup async def get_db(): db = SessionLocal() return db # never closed! # ✅ CORRECT - yield for cleanup async def get_db() -> AsyncGenerator[AsyncSession, None]: async with async_session() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close() @app.get("/items/") async def get_items(db: AsyncSession = Depends(get_db)): # session automatically cleaned up after request pass ``` #### `di-cache` - Cache Expensive Dependencies **Impact**: Reduce redundant computation, faster startup ```python # ❌ INCORRECT - expensive operation every request async def get_settings(): return Settings() # parsed from env every time # ✅ CORRECT - cached dependency (use_cache=True is default) from functools import lru_cache @lru_cache() def get_settings() -> Settings: return Settings() # parsed once, cached # For async dependencies that should be called once per request # but cached within that request (FastAPI default behavior) async def get_current_user(token: str = Depends(oauth2_scheme)): # called only once per request even if used in multiple places return await decode_token(token) ``` #### `di-scope` - Understand Dependency Scopes **Impact**: Correct resource management per lifetime ```python # App-scoped (singleton): created once at startup @lru_cache() def get_redis_pool() -> redis.ConnectionPool: return redis.ConnectionPool(host='localhost', port=6379) # Request-scoped: created per request (default for async generators) async def get_db_session() -> AsyncGenerator[AsyncSession, None]: async with async_session() as session: yield session # No scope (fresh every call): for dependencies with side effects def get_request_id() -> str: return str(uuid.uuid4()) ``` #### `di-override` - Use dependency_overrides for Testing **Impact**: Clean test isolation without modifying code ```python # Production dependency async def get_db() -> AsyncGenerator[AsyncSession, None]: async with async_session() as session: yield session # Test override async def override_get_db() -> AsyncGenerator[AsyncSession, None]: async with test_session() as session: yield session await session.rollback() # rollback for test isolation # In tests def test_get_items(): app.dependency_overrides[get_db] = override_get_db try: client = TestClient(app) response = client.get("/items/") assert response.status_code == 200 finally: app.dependency_overrides.clear() ``` #### `di-annotated` - Use Annotated for Cleaner Signatures **Impact**: Type-safe DI, cleaner function signatures ```python # ❌ INCORRECT - verbose signatures @app.get("/items/") async def get_items( db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), pagination: dict = Depends(pagination) ): pass # ✅ CORRECT - Annotated for reusable dependencies from typing import Annotated DBSession = Annotated[AsyncSession, Depends(get_db)] CurrentUser = Annotated[User, Depends(get_current_user)] Pagination = Annotated[dict, Depends(pagination)] @app.get("/items/") async def get_items( db: DBSession, current_user: CurrentUser, pagination: Pagination ): pass ``` --- ### 3. Pydantic Models (HIGH) #### `pydantic-field` - Use Field() with Constraints **Impact**: Automatic validation, clear API contracts ```python # ❌ INCORRECT - no validation class UserCreate(BaseModel): email: str age: int name: str # ✅ CORRECT - with validation constraints from pydantic import BaseModel, EmailStr, Field class UserCreate(BaseModel): email: EmailStr age: int = Field(ge=0, le=150, description="User age in years") name: str = Field(min_length=1, max_length=100, pattern=r'^[a-zA-Z\s]+$') password: str = Field(min_length=8, max_length=100) model_config = ConfigDict( str_strip_whitespace=True, json_schema_extra={ "example": { "email": "user@example.com", "age": 25, "name": "John Doe", "password": "securepass123" } } ) ``` #### `pydantic-validator` - Use @field_validator for Custom Logic **Impact**: Complex validation, business rules enforcement ```python from pydantic import BaseModel, field_validator, model_validator class OrderCreate(BaseModel): quantity: int price: float discount: float = 0.0 @field_validator('quantity') @classmethod def quantity_must_be_positive(cls, v: int) -> int: if v <= 0: raise ValueError('quantity must be positive') return v @field_validator('discount') @classmethod def discount_in_range(cls, v: float) -> float: if not 0 <= v <= 1: raise ValueError('discount must be between 0 and 1') return v @model_validator(mode='after') def validate_total(self) -> 'OrderCreate': total = self.quantity * self.price * (1 - self.discount) if total < 0: raise ValueError('total cannot be negative') return self ``` #### `pydantic-config` - Configure Model Behavior **Impact**: Performance tuning, serialization control ```python from pydantic import BaseModel, ConfigDict class UserResponse(BaseModel): id: int email: str name: str created_at: datetime model_config = ConfigDict( # ORM mode for SQLAlchemy models from_attributes=True, # Strip whitespace from strings str_strip_whitespace=True, # Use enum values instead of names use_enum_values=True, # Validate on assignment (not just creation) validate_assignment=True, # Forbid extra fields extra='forbid', # Populate by field name, not alias populate_by_name=True, # Serialize datetime as ISO format json_encoders={datetime: lambda v: v.isoformat()} ) ``` #### `pydantic-orm` - Use from_attributes for ORM **Impact**: Clean serialization from SQLAlchemy models ```python # ❌ INCORRECT - manual conversion @app.get("/users/{user_id}") async def get_user(user_id: int, db: DBSession): user = await db.get(User, user_id) return { "id": user.id, "email": user.email, "name": user.name } # ✅ CORRECT - automatic ORM conversion class UserResponse(BaseModel): id: int email: str name: str posts: list['PostResponse'] = [] model_config = ConfigDict(from_attributes=True) @app.get("/users/{user_id}", response_model=UserResponse) async def get_user(user_id: int, db: DBSession): user = await db.get(User, user_id) return user # automatic conversion via from_attributes ``` #### `pydantic-generic` - Use Generic Models for Reuse **Impact**: DRY response models, consistent API format ```python from typing import Generic, TypeVar from pydantic import BaseModel T = TypeVar('T') class PaginatedResponse(BaseModel, Generic[T]): items: list[T] total: int page: int size: int pages: int @classmethod def create(cls, items: list[T], total: int, page: int, size: int): return cls( items=items, total=total, page=page, size=size, pages=(total + size - 1) // size ) class APIResponse(BaseModel, Generic[T]): success: bool = True data: T | None = None error: str | None = None # Usage @app.get("/users/", response_model=PaginatedResponse[UserResponse]) async def list_users(pagination: Pagination, db: DBSession): users, total = await get_users_paginated(db, pagination) return PaginatedResponse.create(users, total, pagination['skip'], pagination['limit']) ``` #### `pydantic-alias` - Use Alias for API Compatibility **Impact**: Clean external APIs, internal naming freedom ```python from pydantic import BaseModel, Field # External API uses camelCase, internal uses snake_case class UserCreate(BaseModel): first_name: str = Field(alias='firstName') last_name: str = Field(alias='lastName') email_address: str = Field(alias='emailAddress') model_config = ConfigDict( populate_by_name=True, # accept both alias and field name json_schema_extra={ "example": { "firstName": "John", "lastName": "Doe", "emailAddress": "john@example.com" } } ) class UserResponse(BaseModel): first_name: str = Field(serialization_alias='firstName') last_name: str = Field(serialization_alias='lastName') model_config = ConfigDict( from_attributes=True, by_alias=True # serialize using alias ) ``` --- ### 4. Database Patterns (HIGH) #### `db-async` - Use Async SQLAlchemy Engine **Impact**: Non-blocking database operations ```python # ❌ INCORRECT - sync engine blocks event loop from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine = create_engine("postgresql://user:pass@localhost/db") SessionLocal = sessionmaker(bind=engine) # ✅ CORRECT - async engine with asyncpg from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker engine = create_async_engine( "postgresql+asyncpg://user:pass@localhost/db", echo=False, pool_size=5, max_overflow=10, pool_pre_ping=True ) async_session = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) ``` #### `db-session` - Use Async Session Context Manager **Impact**: Connection safety, automatic cleanup ```python # ❌ INCORRECT - manual session management async def get_users(): session = async_session() users = await session.execute(select(User)) # session never closed, connection leak! return users.scalars().all() # ✅ CORRECT - async context manager async def get_db() -> AsyncGenerator[AsyncSession, None]: async with async_session() as session: try: yield session await session.commit() except Exception: await session.rollback() raise @app.get("/users/") async def get_users(db: DBSession): result = await db.execute(select(User)) return result.scalars().all() ``` #### `db-n-plus-one` - Avoid N+1 with Eager Loading **Impact**: Query optimization, reduced database round trips ```python # ❌ INCORRECT - N+1 query problem @app.get("/users/") async def get_users_with_posts(db: DBSession): result = await db.execute(select(User)) users = result.scalars().all() for user in users: # This triggers N additional queries! posts = user.posts # lazy load for each user return users # ✅ CORRECT - eager loading with selectinload from sqlalchemy.orm import selectinload, joinedload @app.get("/users/") async def get_users_with_posts(db: DBSession): # selectinload: separate IN query (good for large collections) stmt = select(User).options(selectinload(User.posts)) # joinedload: single JOIN query (good for small collections) # stmt = select(User).options(joinedload(User.posts)) result = await db.execute(stmt) return result.scalars().unique().all() # For nested relationships stmt = select(User).options( selectinload(User.posts).selectinload(Post.comments) ) ``` #### `db-transaction` - Use Explicit Transactions **Impact**: Data consistency, atomicity ```python # ❌ INCORRECT - implicit transaction, partial failure possible @app.post("/transfer/") async def transfer_money(from_id: int, to_id: int, amount: float, db: DBSession): from_account = await db.get(Account, from_id) from_account.balance -= amount await db.commit() # committed to_account = await db.get(Account, to_id) to_account.balance += amount # if this fails, money is lost! await db.commit() # ✅ CORRECT - explicit transaction @app.post("/transfer/") async def transfer_money(from_id: int, to_id: int, amount: float, db: DBSession): async with db.begin(): # explicit transaction from_account = await db.get(Account, from_id) to_account = await db.get(Account, to_id) if from_account.balance < amount: raise HTTPException(400, "Insufficient funds") from_account.balance -= amount to_account.balance += amount # both changes committed together or both rolled back ``` #### `db-migration` - Use Alembic Async Migrations **Impact**: Schema management, version control ```python # alembic/env.py - async migration configuration from alembic import context from sqlalchemy.ext.asyncio import create_async_engine import asyncio def run_migrations_online(): connectable = create_async_engine(settings.database_url) def do_run_migrations_sync(connection): context.configure( connection=connection, target_metadata=Base.metadata, compare_type=True, compare_server_default=True ) with context.begin_transaction(): context.run_migrations() async def run_async_migrations(): async with connectable.connect() as connection: await connection.run_sync(do_run_migrations_sync) await connectable.dispose() asyncio.run(run_async_migrations()) ``` #### `db-pool` - Configure Connection Pool Properly **Impact**: Resource efficiency, connection management ```python from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.pool import AsyncAdaptedQueuePool engine = create_async_engine( "postgresql+asyncpg://user:pass@localhost/db", poolclass=AsyncAdaptedQueuePool, pool_size=5, # base pool size max_overflow=10, # additional connections allowed pool_timeout=30, # wait timeout for connection pool_recycle=1800, # recycle connections after 30 min pool_pre_ping=True, # test connections before use echo=False, # disable SQL logging in production echo_pool="debug", # pool event logging for debugging ) # Graceful shutdown @app.on_event("shutdown") async def shutdown(): await engine.dispose() ``` --- ### 5. Error Handling (MEDIUM) #### `err-http` - Use HTTPException with Status Codes **Impact**: Proper HTTP semantics, clear error messages ```python # ❌ INCORRECT - generic exception @app.get("/items/{item_id}") async def get_item(item_id: int): item = await get_item_from_db(item_id) if not item: raise Exception("Not found") # returns 500 # ✅ CORRECT - HTTPException with proper status from fastapi import HTTPException, status @app.get("/items/{item_id}") async def get_item(item_id: int): item = await get_item_from_db(item_id) if not item: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Item {item_id} not found", headers={"X-Error": "Item not found"} ) return item ``` #### `err-handler` - Register Global Exception Handlers **Impact**: Centralized error handling, consistent format ```python from fastapi import Request from fastapi.responses import JSONResponse class AppError(Exception): def __init__(self, code: str, message: str, status_code: int = 400): self.code = code self.message = message self.status_code = status_code @app.exception_handler(AppError) async def app_error_handler(request: Request, exc: AppError): return JSONResponse( status_code=exc.status_code, content={ "success": False, "error": { "code": exc.code, "message": exc.message } } ) @app.exception_handler(Exception) async def generic_error_handler(request: Request, exc: Exception): logger.exception("Unhandled exception") return JSONResponse( status_code=500, content={ "success": False, "error": { "code": "INTERNAL_ERROR", "message": "An unexpected error occurred" } } ) ``` #### `err-validation` - Handle RequestValidationError **Impact**: User-friendly validation messages ```python from fastapi.exceptions import RequestValidationError from pydantic import ValidationError @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): errors = [] for error in exc.errors(): field = ".".join(str(loc) for loc in error["loc"]) errors.append({ "field": field, "message": error["msg"], "type": error["type"] }) return JSONResponse( status_code=422, content={ "success": False, "error": { "code": "VALIDATION_ERROR", "message": "Request validation failed", "details": errors } } ) ``` #### `err-middleware` - Use Middleware for Cross-Cutting **Impact**: Consistent error format across all endpoints ```python from starlette.middleware.base import BaseHTTPMiddleware import traceback class ErrorHandlingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): try: response = await call_next(request) return response except Exception as exc: logger.error(f"Request failed: {traceback.format_exc()}") return JSONResponse( status_code=500, content={ "success": False, "error": { "code": "INTERNAL_ERROR", "message": str(exc) if settings.DEBUG else "Internal server error", "request_id": request.state.request_id } } ) app.add_middleware(ErrorHandlingMiddleware) ``` #### `err-logging` - Log Errors with Context **Impact**: Debugging support, error tracking ```python import logging import structlog from uuid import uuid4 # Configure structured logging structlog.configure( processors=[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer() ], wrapper_class=structlog.stdlib.BoundLogger, ) logger = structlog.get_logger() @app.middleware("http") async def logging_middleware(request: Request, call_next): request_id = str(uuid4()) request.state.request_id = request_id with structlog.contextvars.bound_contextvars( request_id=request_id, method=request.method, path=request.url.path ): try: response = await call_next(request) logger.info("request_completed", status_code=response.status_code) return response except Exception as exc: logger.exception("request_failed", error=str(exc)) raise ``` #### `err-retry` - Implement Retry with Backoff **Impact**: Resilience for transient failures ```python import asyncio from functools import wraps from typing import Type def retry_async( max_retries: int = 3, backoff_factor: float = 2.0, exceptions: tuple[Type[Exception], ...] = (Exception,) ): def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_retries): try: return await func(*args, **kwargs) except exceptions as exc: last_exception = exc if attempt < max_retries - 1: wait_time = backoff_factor ** attempt logger.warning( f"Retry {attempt + 1}/{max_retries} after {wait_time}s", error=str(exc) ) await asyncio.sleep(wait_time) raise last_exception return wrapper return decorator @retry_async(max_retries=3, exceptions=(httpx.RequestError,)) async def call_external_api(url: str) -> dict: async with httpx.AsyncClient() as client: response = await client.get(url) response.raise_for_status() return response.json() ``` --- ### 6. Security (MEDIUM) #### `sec-oauth2` - Use OAuth2PasswordBearer **Impact**: Standard auth flow, OpenAPI integration ```python from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") @app.post("/token") async def login(form_data: OAuth2PasswordRequestForm = Depends()): user = await authenticate_user(form_data.username, form_data.password) if not user: raise HTTPException( status_code=401, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"} ) access_token = create_access_token(data={"sub": user.username}) return {"access_token": access_token, "token_type": "bearer"} async def get_current_user(token: str = Depends(oauth2_scheme)) -> User: credentials_exception = HTTPException( status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"} ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception except JWTError: raise credentials_exception user = await get_user(username) if user is None: raise credentials_exception return user ``` #### `sec-jwt` - Implement JWT Properly **Impact**: Stateless authentication, secure token handling ```python from datetime import datetime, timedelta from jose import JWTError, jwt from passlib.context import CryptContext SECRET_KEY = settings.SECRET_KEY # from environment ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 REFRESH_TOKEN_EXPIRE_DAYS = 7 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str: to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({ "exp": expire, "iat": datetime.utcnow(), "type": "access" }) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) def create_refresh_token(data: dict) -> str: to_encode = data.copy() expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS) to_encode.update({ "exp": expire, "iat": datetime.utcnow(), "type": "refresh" }) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: return pwd_context.hash(password) ``` #### `sec-cors` - Configure CORS Correctly **Impact**: Cross-origin security, frontend integration ```python from fastapi.middleware.cors import CORSMiddleware # ❌ INCORRECT - overly permissive app.add_middleware( CORSMiddleware, allow_origins=["*"], # allows any origin! allow_methods=["*"], allow_headers=["*"], ) # ✅ CORRECT - restrictive CORS app.add_middleware( CORSMiddleware, allow_origins=[ "https://myapp.com", "https://staging.myapp.com", "http://localhost:3000" if settings.DEBUG else None ], allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"], allow_headers=["Authorization", "Content-Type", "X-Request-ID"], expose_headers=["X-Total-Count", "X-Page-Count"], max_age=600, # cache preflight for 10 minutes ) ``` #### `sec-rate-limit` - Implement Rate Limiting **Impact**: DoS protection, resource management ```python from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Per-route rate limiting @app.get("/api/items/") @limiter.limit("100/minute") async def get_items(request: Request): return items # Per-user rate limiting def get_user_identifier(request: Request) -> str: if hasattr(request.state, "user"): return f"user:{request.state.user.id}" return get_remote_address(request) @app.post("/api/expensive-operation/") @limiter.limit("10/hour", key_func=get_user_identifier) async def expensive_operation(request: Request): pass ``` #### `sec-input` - Validate and Sanitize Input **Impact**: Injection prevention, data integrity ```python import re import bleach from pydantic import field_validator class CommentCreate(BaseModel): content: str = Field(max_length=10000) @field_validator('content') @classmethod def sanitize_content(cls, v: str) -> str: # Remove potentially dangerous HTML return bleach.clean( v, tags=['p', 'br', 'strong', 'em', 'a'], attributes={'a': ['href']}, protocols=['http', 'https'] ) class SearchQuery(BaseModel): q: str = Field(max_length=200) @field_validator('q') @classmethod def validate_query(cls, v: str) -> str: # Remove SQL injection patterns dangerous_patterns = [ r"'.*--", r";\s*DROP", r";\s*DELETE", r"UNION\s+SELECT" ] for pattern in dangerous_patterns: if re.search(pattern, v, re.IGNORECASE): raise ValueError("Invalid search query") return v.strip() ``` #### `sec-headers` - Set Security Headers **Impact**: Browser protection, security hardening ```python from starlette.middleware.base import BaseHTTPMiddleware class SecurityHeadersMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): response = await call_next(request) # Prevent clickjacking response.headers["X-Frame-Options"] = "DENY" # Prevent MIME type sniffing response.headers["X-Content-Type-Options"] = "nosniff" # Enable XSS protection response.headers["X-XSS-Protection"] = "1; mode=block" # Referrer policy response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" # Content Security Policy response.headers["Content-Security-Policy"] = ( "default-src 'self'; " "script-src 'self'; " "style-src 'self' 'unsafe-inline'; " "img-src 'self' data: https:; " ) # HSTS (only in production) if not settings.DEBUG: response.headers["Strict-Transport-Security"] = ( "max-age=31536000; includeSubDomains" ) return response app.add_middleware(SecurityHeadersMiddleware) ``` --- ### 7. Performance (MEDIUM) #### `perf-response` - Use response_model for Filtering **Impact**: Reduce payload size, prevent data leaks ```python # ❌ INCORRECT - returns entire ORM object @app.get("/users/{user_id}") async def get_user(user_id: int, db: DBSession): return await db.get(User, user_id) # includes password_hash! # ✅ CORRECT - filtered response class UserResponse(BaseModel): id: int email: str name: str # password_hash not included! model_config = ConfigDict(from_attributes=True) @app.get("/users/{user_id}", response_model=UserResponse) async def get_user(user_id: int, db: DBSession): user = await db.get(User, user_id) return user # only fields in UserResponse are serialized # For excluding specific fields dynamically @app.get("/users/{user_id}") async def get_user( user_id: int, db: DBSession, include_email: bool = False ) -> UserResponse: user = await db.get(User, user_id) response = UserResponse.model_validate(user) if not include_email: response.email = None return response ``` #### `perf-cache` - Implement Caching Strategies **Impact**: Reduce database load, faster responses ```python import redis.asyncio as redis from functools import wraps import json redis_client = redis.from_url("redis://localhost:6379") def cache_response(ttl: int = 300, key_prefix: str = "cache"): def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): # Generate cache key from function name and arguments cache_key = f"{key_prefix}:{func.__name__}:{hash(str(args) + …(truncated)