Best Practices

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

Quick Reference - All 48 Rules

Async Patterns (CRITICAL) - async-*

Dependency Injection (CRITICAL) - di-*

Pydantic Models (HIGH) - pydantic-*

Database Patterns (HIGH) - db-*

Error Handling (MEDIUM) - err-*

Security (MEDIUM) - sec-*

Performance (MEDIUM) - perf-*

Testing (LOW) - test-*

Detailed Rules by Category

1. Async Patterns (CRITICAL)

async-block - Never Block the Event Loop

Impact: Server freeze, request timeouts, degraded throughput

# ❌ 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

# ❌ 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

# ❌ 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

# ❌ 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

# ❌ 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

# ❌ 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

# ❌ 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

# ❌ 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

# ❌ 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

# 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

# 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

# ❌ 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

# ❌ 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

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

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

# ❌ 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

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

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

# ❌ 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

# ❌ 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

# ❌ 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

# ❌ 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

# 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

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

# ❌ 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

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

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

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

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

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

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

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

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

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

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

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

# ❌ 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

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)