Python Fastapi

Quick Reference

When to Use This Skill

Use for FastAPI development:

• Setting up FastAPI project structure
• Creating Pydantic schemas with validation
• Implementing dependency injection
• JWT authentication setup
• Docker deployment configuration

Related skills:

• For async patterns: see python-asyncio
• For testing: see python-testing
• For type hints: see python-type-hints

---

FastAPI Production Best Practices (2025)

Overview

FastAPI is a modern, high-performance web framework for building APIs. Built on Starlette and Pydantic, it provides automatic validation, serialization, and OpenAPI documentation.

Project Structure

Scalable Structure (Recommended)

src/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application factory
│   ├── config.py            # Settings management
│   ├── database.py          # Database setup
│   ├── dependencies.py      # Shared dependencies
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── router.py    # API router
│   │   │   └── endpoints/
│   │   │       ├── users.py
│   │   │       ├── items.py
│   │   │       └── auth.py
│   │   └── deps.py          # API dependencies
│   ├── core/
│   │   ├── __init__.py
│   │   ├── security.py      # Auth/JWT
│   │   └── exceptions.py    # Custom exceptions
│   ├── models/              # SQLAlchemy models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/             # Pydantic schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   └── services/            # Business logic
│       ├── __init__.py
│       ├── user.py
│       └── item.py
├── tests/
├── pyproject.toml
└── Dockerfile

Configuration

Settings with pydantic-settings

# app/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    # Application
    app_name: str = "My API"
    debug: bool = False
    environment: str = "production"

    # Database
    database_url: str
    database_pool_size: int = 5
    database_max_overflow: int = 10

    # Security
    secret_key: str
    access_token_expire_minutes: int = 30
    algorithm: str = "HS256"

    # External services
    redis_url: str | None = None


@lru_cache
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

Application Factory

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.api.v1.router import api_router
from app.database import engine, Base


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown
    await engine.dispose()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.app_name,
        openapi_url="/api/v1/openapi.json" if settings.debug else None,
        docs_url="/api/docs" if settings.debug else None,
        redoc_url="/api/redoc" if settings.debug else None,
        lifespan=lifespan,
    )

    # Middleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"] if settings.debug else ["https://myapp.com"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Routes
    app.include_router(api_router, prefix="/api/v1")

    return app


app = create_app()

Pydantic Schemas

Input/Output Separation

# app/schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field

# Base schema with shared fields
class UserBase(BaseModel):
    email: EmailStr
    full_name: str | None = None

# Input schema for creation
class UserCreate(UserBase):
    password: str = Field(min_length=8)

# Input schema for updates
class UserUpdate(BaseModel):
    email: EmailStr | None = None
    full_name: str | None = None
    password: str | None = Field(None, min_length=8)

# Output schema (what API returns)
class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}

# Internal schema with password hash
class UserInDB(UserBase):
    id: int
    hashed_password: str
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}

Validation Examples

from typing import Annotated
from pydantic import BaseModel, Field, field_validator, model_validator

class ItemCreate(BaseModel):
    name: Annotated[str, Field(min_length=1, max_length=100)]
    price: Annotated[float, Field(gt=0, description="Price must be positive")]
    quantity: Annotated[int, Field(ge=0, le=10000)]
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be empty or whitespace")
        return v.strip()

    @field_validator("tags")
    @classmethod
    def tags_must_be_unique(cls, v: list[str]) -> list[str]:
        if len(v) != len(set(v)):
            raise ValueError("Tags must be unique")
        return v

    @model_validator(mode="after")
    def validate_price_quantity(self) -> "ItemCreate":
        if self.quantity > 0 and self.price < 0.01:
            raise ValueError("Price too low for available items")
        return self

Async Best Practices

Correct Async Usage

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
import httpx

app = FastAPI()

# GOOD: Async database operations
async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# GOOD: Async HTTP calls
async def fetch_external_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

# BAD: Blocking call in async route
@app.get("/bad")
async def bad_endpoint():
    import time
    time.sleep(5)  # Blocks entire event loop!
    return {"status": "done"}

# GOOD: Use asyncio.sleep or run in executor
@app.get("/good")
async def good_endpoint():
    import asyncio
    await asyncio.sleep(5)  # Non-blocking
    return {"status": "done"}

# GOOD: Run blocking code in executor
@app.get("/cpu-bound")
async def cpu_bound():
    import asyncio
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, heavy_computation)
    return {"result": result}

Sync vs Async Routes

# Async route - for I/O-bound operations
@app.get("/async-users")
async def get_users(db: AsyncSession = Depends(get_db)):
    # Uses async database session
    users = await db.execute(select(User))
    return users.scalars().all()

# Sync route - FastAPI runs in thread pool
@app.get("/sync-compute")
def compute_heavy():
    # OK for CPU-bound operations
    # FastAPI automatically runs this in thread pool
    return {"result": heavy_cpu_computation()}

Dependencies

Dependency Injection

# app/api/deps.py
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from jose import JWTError, jwt

from app.database import get_db
from app.config import settings
from app.models import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await db.get(User, user_id)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# Type aliases for cleaner signatures
CurrentUser = Annotated[User, Depends(get_current_active_user)]
DbSession = Annotated[AsyncSession, Depends(get_db)]

Using Dependencies

# app/api/v1/endpoints/users.py
from fastapi import APIRouter, HTTPException
from app.api.deps import CurrentUser, DbSession
from app.schemas.user import UserResponse, UserUpdate

router = APIRouter()

@router.get("/me", response_model=UserResponse)
async def read_current_user(current_user: CurrentUser):
    return current_user

@router.patch("/me", response_model=UserResponse)
async def update_current_user(
    update_data: UserUpdate,
    current_user: CurrentUser,
    db: DbSession,
):
    for field, value in update_data.model_dump(exclude_unset=True).items():
        setattr(current_user, field, value)
    db.add(current_user)
    await db.commit()
    await db.refresh(current_user)
    return current_user

Error Handling

Custom Exception Handlers

# app/core/exceptions.py
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from starlette.status import HTTP_500_INTERNAL_SERVER_ERROR

class AppException(Exception):
    def __init__(self, status_code: int, detail: str, error_code: str = None):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code

class NotFoundError(AppException):
    def __init__(self, resource: str, id: int):
        super().__init__(
            status_code=404,
            detail=f"{resource} with id {id} not found",
            error_code="NOT_FOUND",
        )

class ValidationError(AppException):
    def __init__(self, detail: str):
        super().__init__(
            status_code=422,
            detail=detail,
            error_code="VALIDATION_ERROR",
        )

# app/main.py
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "detail": exc.detail,
            "error_code": exc.error_code,
        },
    )

async def unhandled_exception_handler(request: Request, exc: Exception):
    # Log the error
    logger.exception(f"Unhandled error: {exc}")
    return JSONResponse(
        status_code=HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal server error"},
    )

app.add_exception_handler(AppException, app_exception_handler)
app.add_exception_handler(Exception, unhandled_exception_handler)

Production Deployment

Docker

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# Install uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv

# Copy dependency files
COPY pyproject.toml uv.lock ./

# Install dependencies
RUN uv sync --frozen --no-dev

# Copy application
COPY src/ ./src/

# Run with uvicorn
CMD ["uv", "run", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Gunicorn with Uvicorn Workers

# gunicorn.conf.py
import multiprocessing

# Workers = CPU cores (for async)
workers = multiprocessing.cpu_count()
worker_class = "uvicorn.workers.UvicornWorker"

# Timeouts
timeout = 120
keepalive = 5

# Logging
accesslog = "-"
errorlog = "-"
loglevel = "info"

# Binding
bind = "0.0.0.0:8000"

# Run in production
gunicorn app.main:app -c gunicorn.conf.py

Health Checks

from fastapi import APIRouter

router = APIRouter(tags=["health"])

@router.get("/health")
async def health_check():
    return {"status": "healthy"}

@router.get("/ready")
async def readiness_check(db: DbSession):
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "ready", "database": "connected"}
    except Exception:
        raise HTTPException(status_code=503, detail="Database not ready")

Security

JWT Authentication

# app/core/security.py
from datetime import datetime, timedelta
from jose import jwt
from passlib.context import CryptContext
from app.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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)

def create_access_token(subject: int | str, expires_delta: timedelta = None) -> str:
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.access_token_expire_minutes)

    to_encode = {"exp": expire, "sub": str(subject)}
    return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)

Rate Limiting

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)

@app.get("/limited")
@limiter.limit("10/minute")
async def limited_endpoint(request: Request):
    return {"message": "This endpoint is rate limited"}

Testing

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from app.main import app
from app.database import get_db

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        yield client

@pytest.fixture
async def authenticated_client(client: AsyncClient, test_user_token: str):
    client.headers["Authorization"] = f"Bearer {test_user_token}"
    yield client

# tests/test_users.py
import pytest

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/api/v1/users", json={
        "email": "test@example.com",
        "password": "testpassword123",
    })
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"

Additional References

For advanced patterns beyond this guide, see:

• [Dependency Injection Patterns](references/dependency-injection-patterns.md) - Type aliases (DbSession, CurrentUser, AdminUser), pagination dependencies, sorting/filtering, service layer patterns, request context, feature flags, rate limiting, caching dependencies

FastAPI Dependency Injection Patterns

Advanced dependency injection patterns for production FastAPI applications.

Type Aliases for Clean Signatures

from typing import Annotated
from fastapi import Depends, Query, Header
from sqlalchemy.ext.asyncio import AsyncSession

# Database session
async def get_db() -> AsyncIterator[AsyncSession]:
    async with async_session() as session:
        yield session

DbSession = Annotated[AsyncSession, Depends(get_db)]

# Current user
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: DbSession,
) -> User:
    user = await authenticate_user(token, db)
    if not user:
        raise HTTPException(401, "Invalid token")
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

# Active user (verified, not banned)
async def get_active_user(user: CurrentUser) -> User:
    if not user.is_active:
        raise HTTPException(403, "User is inactive")
    return user

ActiveUser = Annotated[User, Depends(get_active_user)]

# Admin user
async def get_admin_user(user: ActiveUser) -> User:
    if not user.is_admin:
        raise HTTPException(403, "Admin access required")
    return user

AdminUser = Annotated[User, Depends(get_admin_user)]


# Usage - clean endpoint signatures
@router.get("/users/me")
async def get_me(user: CurrentUser) -> UserResponse:
    return user

@router.delete("/users/{user_id}")
async def delete_user(user_id: int, admin: AdminUser, db: DbSession) -> None:
    await db.execute(delete(User).where(User.id == user_id))
    await db.commit()

Pagination Dependency

from dataclasses import dataclass
from typing import Annotated, Generic, TypeVar
from fastapi import Query

T = TypeVar("T")

@dataclass
class PaginationParams:
    page: int
    size: int
    offset: int

    @property
    def limit(self) -> int:
        return self.size


def get_pagination(
    page: Annotated[int, Query(ge=1, description="Page number")] = 1,
    size: Annotated[int, Query(ge=1, le=100, description="Items per page")] = 20,
) -> PaginationParams:
    return PaginationParams(
        page=page,
        size=size,
        offset=(page - 1) * size,
    )

Pagination = Annotated[PaginationParams, Depends(get_pagination)]


@dataclass
class PagedResponse(Generic[T]):
    items: list[T]
    total: int
    page: int
    size: int
    pages: int


# Usage
@router.get("/users", response_model=PagedResponse[UserResponse])
async def list_users(
    db: DbSession,
    pagination: Pagination,
) -> PagedResponse[UserResponse]:
    # Count total
    total = await db.scalar(select(func.count()).select_from(User))

    # Get page
    query = select(User).offset(pagination.offset).limit(pagination.limit)
    result = await db.execute(query)
    users = result.scalars().all()

    return PagedResponse(
        items=users,
        total=total,
        page=pagination.page,
        size=pagination.size,
        pages=(total + pagination.size - 1) // pagination.size,
    )

Sorting and Filtering

from enum import Enum
from typing import Annotated
from fastapi import Query

class SortOrder(str, Enum):
    ASC = "asc"
    DESC = "desc"

@dataclass
class SortParams:
    field: str
    order: SortOrder

def get_sort(
    sort_by: Annotated[str, Query(description="Field to sort by")] = "created_at",
    order: Annotated[SortOrder, Query(description="Sort order")] = SortOrder.DESC,
) -> SortParams:
    return SortParams(field=sort_by, order=order)

Sort = Annotated[SortParams, Depends(get_sort)]


@dataclass
class FilterParams:
    search: str | None
    status: str | None
    created_after: datetime | None
    created_before: datetime | None

def get_filters(
    search: Annotated[str | None, Query(description="Search term")] = None,
    status: Annotated[str | None, Query(description="Status filter")] = None,
    created_after: Annotated[datetime | None, Query()] = None,
    created_before: Annotated[datetime | None, Query()] = None,
) -> FilterParams:
    return FilterParams(
        search=search,
        status=status,
        created_after=created_after,
        created_before=created_before,
    )

Filters = Annotated[FilterParams, Depends(get_filters)]


# Usage
@router.get("/items")
async def list_items(
    db: DbSession,
    pagination: Pagination,
    sort: Sort,
    filters: Filters,
) -> PagedResponse[ItemResponse]:
    query = select(Item)

    # Apply filters
    if filters.search:
        query = query.where(Item.name.ilike(f"%{filters.search}%"))
    if filters.status:
        query = query.where(Item.status == filters.status)
    if filters.created_after:
        query = query.where(Item.created_at >= filters.created_after)
    if filters.created_before:
        query = query.where(Item.created_at <= filters.created_before)

    # Apply sorting
    sort_column = getattr(Item, sort.field, Item.created_at)
    if sort.order == SortOrder.DESC:
        query = query.order_by(sort_column.desc())
    else:
        query = query.order_by(sort_column.asc())

    # Apply pagination
    query = query.offset(pagination.offset).limit(pagination.limit)

    result = await db.execute(query)
    return result.scalars().all()

Service Layer Dependencies

from typing import Protocol, Annotated
from fastapi import Depends

# Service protocol
class UserServiceProtocol(Protocol):
    async def get_by_id(self, user_id: int) -> User | None: ...
    async def create(self, data: UserCreate) -> User: ...
    async def update(self, user_id: int, data: UserUpdate) -> User: ...
    async def delete(self, user_id: int) -> bool: ...


# Service implementation
class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        return await self.db.get(User, user_id)

    async def create(self, data: UserCreate) -> User:
        user = User(**data.model_dump(exclude={"password"}))
        user.hashed_password = hash_password(data.password)
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def update(self, user_id: int, data: UserUpdate) -> User:
        user = await self.get_by_id(user_id)
        if not user:
            raise HTTPException(404, "User not found")
        for field, value in data.model_dump(exclude_unset=True).items():
            setattr(user, field, value)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def delete(self, user_id: int) -> bool:
        result = await self.db.execute(
            delete(User).where(User.id == user_id)
        )
        await self.db.commit()
        return result.rowcount > 0


# Dependency
def get_user_service(db: DbSession) -> UserService:
    return UserService(db)

UserSvc = Annotated[UserService, Depends(get_user_service)]


# Usage
@router.get("/users/{user_id}")
async def get_user(user_id: int, service: UserSvc) -> UserResponse:
    user = await service.get_by_id(user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

@router.post("/users", status_code=201)
async def create_user(data: UserCreate, service: UserSvc) -> UserResponse:
    return await service.create(data)

Request Context / Scope Dependency

from contextvars import ContextVar
from uuid import uuid4
from fastapi import Request

# Context variables
request_id_ctx: ContextVar[str] = ContextVar("request_id", default="")
user_ctx: ContextVar[User | None] = ContextVar("current_user", default=None)


# Middleware to set request context
@app.middleware("http")
async def request_context_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid4()))
    request_id_ctx.set(request_id)

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response


# Access context anywhere
def get_request_id() -> str:
    return request_id_ctx.get()

RequestId = Annotated[str, Depends(get_request_id)]


# Usage in logging
import structlog

def get_logger() -> structlog.BoundLogger:
    return structlog.get_logger().bind(request_id=request_id_ctx.get())

Logger = Annotated[structlog.BoundLogger, Depends(get_logger)]


@router.post("/orders")
async def create_order(
    data: OrderCreate,
    db: DbSession,
    user: CurrentUser,
    request_id: RequestId,
    logger: Logger,
) -> OrderResponse:
    logger.info("Creating order", user_id=user.id)
    # ... create order

Feature Flags Dependency

from dataclasses import dataclass

@dataclass
class FeatureFlags:
    new_dashboard: bool
    beta_api: bool
    dark_mode: bool


async def get_feature_flags(user: CurrentUser | None = None) -> FeatureFlags:
    """Get feature flags for current user."""
    # Could fetch from database, config, or feature flag service
    base_flags = FeatureFlags(
        new_dashboard=True,
        beta_api=False,
        dark_mode=True,
    )

    # User-specific overrides
    if user and user.is_beta_tester:
        base_flags.beta_api = True

    return base_flags

Features = Annotated[FeatureFlags, Depends(get_feature_flags)]


@router.get("/dashboard")
async def get_dashboard(features: Features) -> dict:
    if features.new_dashboard:
        return {"version": "v2", "data": [...]}
    return {"version": "v1", "data": [...]}

Rate Limiting Dependency

from fastapi import HTTPException, Request
from dataclasses import dataclass
import redis.asyncio as redis

@dataclass
class RateLimitResult:
    allowed: bool
    remaining: int
    reset_at: int


class RateLimiter:
    def __init__(self, redis_client: redis.Redis, limit: int, window: int):
        self.redis = redis_client
        self.limit = limit
        self.window = window

    async def check(self, key: str) -> RateLimitResult:
        now = int(time.time())
        window_start = now - self.window

        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(key, 0, window_start)
        pipe.zadd(key, {str(now): now})
        pipe.zcard(key)
        pipe.expire(key, self.window)
        _, _, count, _ = await pipe.execute()

        return RateLimitResult(
            allowed=count <= self.limit,
            remaining=max(0, self.limit - count),
            reset_at=now + self.window,
        )


def rate_limit(limit: int = 100, window: int = 60):
    """Rate limit dependency factory."""
    async def dependency(request: Request) -> None:
        limiter = RateLimiter(request.app.state.redis, limit, window)
        key = f"rate_limit:{request.client.host}:{request.url.path}"

        result = await limiter.check(key)
        if not result.allowed:
            raise HTTPException(
                status_code=429,
                detail="Too many requests",
                headers={
                    "X-RateLimit-Remaining": str(result.remaining),
                    "X-RateLimit-Reset": str(result.reset_at),
                },
            )

    return Depends(dependency)


# Usage
@router.post("/api/expensive-operation", dependencies=[rate_limit(10, 60)])
async def expensive_operation() -> dict:
    # Limited to 10 requests per minute
    return {"status": "ok"}

Caching Dependency

import hashlib
import json
from typing import Callable

def cached(ttl: int = 300, key_prefix: str = ""):
    """Cache decorator for endpoints."""
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, request: Request, **kwargs):
            # Generate cache key
            key_data = {
                "path": request.url.path,
                "query": str(request.query_params),
                "prefix": key_prefix,
            }
            cache_key = hashlib.md5(
                json.dumps(key_data, sort_keys=True).encode()
            ).hexdigest()

            # Check cache
            redis = request.app.state.redis
            cached = await redis.get(cache_key)
            if cached:
                return json.loads(cached)

            # Execute and cache
            result = await func(*args, request=request, **kwargs)
            await redis.setex(cache_key, ttl, json.dumps(result))
            return result

        return wrapper
    return decorator


# Usage
@router.get("/products")
@cached(ttl=60, key_prefix="products")
async def list_products(request: Request, db: DbSession) -> list[ProductResponse]:
    result = await db.execute(select(Product))
    return result.scalars().all()