fastapi-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

FastAPI Patterns

FastAPI 模式实践

When to Use

适用场景

Activate this skill when:

Configuring FastAPI middleware (CORS, logging, timing, error handling)
Creating complex dependency injection chains
Implementing WebSocket endpoints with connection management
Customizing OpenAPI documentation (tags, examples, deprecation)
Setting up JWT authentication and role-based access dependencies
Implementing background tasks (lightweight or distributed)
Managing application lifecycle (startup/shutdown via lifespan)
Setting up rate limiting or request throttling

Do NOT use this skill for:

Basic endpoint CRUD, repository, or service patterns (use
```
python-backend-expert
```
)
Writing tests for FastAPI endpoints (use
```
pytest-patterns
```
)
API contract design or schema planning (use
```
api-design-patterns
```
)
Architecture decisions (use
```
system-architecture
```
)

在以下场景中启用此技能：

配置FastAPI中间件（CORS、日志、计时、错误处理）
创建复杂的依赖注入链
实现带连接管理的WebSocket端点
自定义OpenAPI文档（标签、示例、弃用标记）
配置JWT认证与基于角色的访问依赖
实现后台任务（轻量型或分布式）
管理应用生命周期（通过lifespan处理启动/关闭）
配置请求限流或节流

以下场景不适用此技能：

基础端点CRUD、仓储或服务模式（请使用
```
python-backend-expert
```
）
为FastAPI端点编写测试（请使用
```
pytest-patterns
```
）
API契约设计或Schema规划（请使用
```
api-design-patterns
```
）
架构决策（请使用
```
system-architecture
```
）

Instructions

操作指南

Middleware Stack

中间件栈

Middleware executes in LIFO (Last In, First Out) order. The last middleware added is the outermost layer.

python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

中间件遵循LIFO（后进先出）顺序执行。最后添加的中间件是最外层。

python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

Order matters: added last = executed first (outermost)

app.add_middleware(TimingMiddleware) # 3rd added = runs 1st app.add_middleware(RequestLoggingMiddleware) # 2nd added = runs 2nd app.add_middleware( # 1st added = runs 3rd (innermost) CORSMiddleware, allow_origins=["https://app.example.com"], # NEVER use "*" in production allow_credentials=True, allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"], allow_headers=["Authorization", "Content-Type"], )

undefined

undefined

ASGI Middleware (Preferred)

ASGI中间件（推荐使用）

Use pure ASGI middleware for performance-critical paths:

python

from starlette.types import ASGIApp, Receive, Scope, Send
import time

class TimingMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start = time.perf_counter()
        await self.app(scope, receive, send)
        duration = time.perf_counter() - start
        # Log or record the duration

在性能敏感路径中使用纯ASGI中间件：

python

from starlette.types import ASGIApp, Receive, Scope, Send
import time

class TimingMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start = time.perf_counter()
        await self.app(scope, receive, send)
        duration = time.perf_counter() - start
        # Log or record the duration

BaseHTTPMiddleware (Simpler but Slower)

BaseHTTPMiddleware（更简单但性能较低）

Use only for middleware that needs to read/modify the request body or response:

python

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

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

When to use which:

ASGI middleware: Performance-critical, no need to read request/response body
BaseHTTPMiddleware: Need access to
```
Request
```
/
```
Response
```
objects, simpler API

仅在需要读取/修改请求体或响应体时使用：

python

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

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

选择指南：

ASGI中间件： 性能敏感场景，无需读取请求/响应体
BaseHTTPMiddleware： 需要访问
```
Request
```
/
```
Response
```
对象，API更简单

Authentication Dependencies

认证依赖

JWT Token Validation

JWT令牌验证

python

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user

python

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user

Role-Based Access (Factory Pattern)

基于角色的访问（工厂模式）

python

def require_role(*roles: str):
    """Factory that creates a dependency requiring specific roles."""
    async def check_role(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(
                status_code=403,
                detail=f"Requires one of: {', '.join(roles)}",
            )
        return user
    return check_role

python

def require_role(*roles: str):
    """Factory that creates a dependency requiring specific roles."""
    async def check_role(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(
                status_code=403,
                detail=f"Requires one of: {', '.join(roles)}",
            )
        return user
    return check_role

Usage in routes

@router.delete("/users/{user_id}", dependencies=[Depends(require_role("admin"))]) async def delete_user(user_id: int, ...) -> None: ...

@router.patch("/posts/{post_id}") async def update_post( post_id: int, user: User = Depends(require_role("admin", "editor")), ) -> PostResponse: ...

undefined

@router.delete("/users/{user_id}", dependencies=[Depends(require_role("admin"))]) async def delete_user(user_id: int, ...) -> None: ...

@router.patch("/posts/{post_id}") async def update_post( post_id: int, user: User = Depends(require_role("admin", "editor")), ) -> PostResponse: ...

undefined

Dependency Injection Chains

依赖注入链

Caching Behavior

缓存行为

FastAPI caches dependency results within a single request. The same dependency called multiple times returns the same instance:

python

undefined

FastAPI会在单个请求内缓存依赖的结果。同一个依赖被多次调用时会返回同一个实例：

python

undefined

get_async_session is called once per request, even if used by multiple deps

async def get_user_service(session: AsyncSession = Depends(get_async_session)) -> UserService: return UserService(session)

async def get_post_service(session: AsyncSession = Depends(get_async_session)) -> PostService: return PostService(session) # Same session instance as user_service


To disable caching (get a new instance each time), use `use_cache=False`:
```python
session: AsyncSession = Depends(get_async_session, use_cache=False)

async def get_user_service(session: AsyncSession = Depends(get_async_session)) -> UserService: return UserService(session)

async def get_post_service(session: AsyncSession = Depends(get_async_session)) -> PostService: return PostService(session) # Same session instance as user_service


要禁用缓存（每次调用获取新实例），请使用`use_cache=False`：
```python
session: AsyncSession = Depends(get_async_session, use_cache=False)

Yield Dependencies (Resource Cleanup)

Yield依赖（资源清理）

python

async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client
    # Client is automatically closed after the request

python

async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client
    # Client is automatically closed after the request

Overriding Dependencies in Tests

测试中覆盖依赖

python

undefined

python

undefined

In tests

from app.main import app from app.dependencies.auth import get_current_user

async def mock_current_user() -> User: return User(id=1, email="test@example.com", role="admin")

app.dependency_overrides[get_current_user] = mock_current_user

undefined

from app.main import app from app.dependencies.auth import get_current_user

async def mock_current_user() -> User: return User(id=1, email="test@example.com", role="admin")

app.dependency_overrides[get_current_user] = mock_current_user

undefined

Background Tasks

后台任务

FastAPI BackgroundTasks (Lightweight)

FastAPI BackgroundTasks（轻量型）

For tasks that don't need to survive server restarts:

python

from fastapi import BackgroundTasks

@router.post("/users", status_code=201)
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    user = await service.create_user(data)
    background_tasks.add_task(send_welcome_email, user.email)
    return UserResponse.model_validate(user)

async def send_welcome_email(email: str) -> None:
    """Runs after the response is sent. Creates its own session."""
    async with async_session_factory() as session:
        async with session.begin():
            # Send email, log activity, etc.
            ...

Rules:

Never reuse the request session in background tasks — create a new one
Background tasks run in the same process — no retry, no persistence
Use Celery or similar for tasks that need reliability, retry, or distribution

适用于不需要在服务器重启后继续执行的任务：

python

from fastapi import BackgroundTasks

@router.post("/users", status_code=201)
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    user = await service.create_user(data)
    background_tasks.add_task(send_welcome_email, user.email)
    return UserResponse.model_validate(user)

async def send_welcome_email(email: str) -> None:
    """Runs after the response is sent. Creates its own session."""
    async with async_session_factory() as session:
        async with session.begin():
            # Send email, log activity, etc.
            ...

规则：

切勿在后台任务中复用请求会话——请创建新会话
后台任务在同一进程中运行——无重试、无持久化
对于需要可靠性、重试或分布式的任务，请使用Celery等工具

WebSocket Pattern

WebSocket模式

python

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self) -> None:
        self.active: dict[int, list[WebSocket]] = {}

    async def connect(self, user_id: int, ws: WebSocket) -> None:
        await ws.accept()
        self.active.setdefault(user_id, []).append(ws)

    def disconnect(self, user_id: int, ws: WebSocket) -> None:
        if user_id in self.active:
            self.active[user_id].remove(ws)
            if not self.active[user_id]:
                del self.active[user_id]

    async def send_to_user(self, user_id: int, message: dict) -> None:
        for ws in self.active.get(user_id, []):
            await ws.send_json(message)

manager = ConnectionManager()

@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket, token: str) -> None:
    # Auth via query parameter: /ws?token=xxx
    user = await verify_ws_token(token)
    if not user:
        await ws.close(code=4001)
        return

    await manager.connect(user.id, ws)
    try:
        while True:
            data = await ws.receive_json()
            # Process incoming messages
            await handle_message(user.id, data)
    except WebSocketDisconnect:
        manager.disconnect(user.id, ws)

WebSocket auth approaches:

Query parameter:
```
ws://host/ws?token=xxx
```
(simplest, token visible in logs)
First message: Connect, then send token as first message (more secure)
Cookie: Use existing session cookie (requires same domain)

python

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self) -> None:
        self.active: dict[int, list[WebSocket]] = {}

    async def connect(self, user_id: int, ws: WebSocket) -> None:
        await ws.accept()
        self.active.setdefault(user_id, []).append(ws)

    def disconnect(self, user_id: int, ws: WebSocket) -> None:
        if user_id in self.active:
            self.active[user_id].remove(ws)
            if not self.active[user_id]:
                del self.active[user_id]

    async def send_to_user(self, user_id: int, message: dict) -> None:
        for ws in self.active.get(user_id, []):
            await ws.send_json(message)

manager = ConnectionManager()

@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket, token: str) -> None:
    # Auth via query parameter: /ws?token=xxx
    user = await verify_ws_token(token)
    if not user:
        await ws.close(code=4001)
        return

    await manager.connect(user.id, ws)
    try:
        while True:
            data = await ws.receive_json()
            # Process incoming messages
            await handle_message(user.id, data)
    except WebSocketDisconnect:
        manager.disconnect(user.id, ws)

WebSocket认证方式：

查询参数：
```
ws://host/ws?token=xxx
```
（最简单，令牌会出现在日志中）
首条消息：先连接，再发送令牌作为首条消息（更安全）
Cookie：使用现有会话Cookie（要求同域）

Application Lifespan

应用生命周期

Use the

lifespan

context manager (not deprecated

on_event

python

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Startup: initialize resources
    await init_database()
    redis = await aioredis.from_url(settings.redis_url)
    app.state.redis = redis

    yield  # Application runs here

    # Shutdown: cleanup resources
    await redis.close()
    await dispose_engine()

app = FastAPI(lifespan=lifespan)

Lifespan responsibilities:

Database connection pool initialization and disposal
Redis/cache connection setup and teardown
HTTP client pool creation
Background scheduler startup/shutdown
Cache warmup on startup

使用

lifespan

上下文管理器（而非已弃用的

on_event

）：

python

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Startup: initialize resources
    await init_database()
    redis = await aioredis.from_url(settings.redis_url)
    app.state.redis = redis

    yield  # Application runs here

    # Shutdown: cleanup resources
    await redis.close()
    await dispose_engine()

app = FastAPI(lifespan=lifespan)

Lifespan职责：

数据库连接池的初始化与销毁
Redis/缓存连接的建立与关闭
HTTP客户端池的创建
后台调度器的启动/关闭
启动时预热缓存

Exception Handlers

异常处理器

python

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

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "code": "VALIDATION_ERROR",
            "field_errors": [
                {"field": e["loc"][-1], "message": e["msg"], "code": e["type"]}
                for e in exc.errors()
            ],
        },
    )

注册全局异常处理器以实现统一的错误响应：

python

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

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "code": "VALIDATION_ERROR",
            "field_errors": [
                {"field": e["loc"][-1], "message": e["msg"], "code": e["type"]}
                for e in exc.errors()
            ],
        },
    )

OpenAPI Customization

OpenAPI自定义

python

app = FastAPI(
    title="My API",
    version="1.0.0",
    description="API description with **markdown** support",
    openapi_tags=[
        {"name": "Users", "description": "User management operations"},
        {"name": "Auth", "description": "Authentication endpoints"},
    ],
    docs_url="/docs",        # Swagger UI
    redoc_url="/redoc",      # ReDoc
    openapi_url="/openapi.json",
)

python

app = FastAPI(
    title="My API",
    version="1.0.0",
    description="API description with **markdown** support",
    openapi_tags=[
        {"name": "Users", "description": "User management operations"},
        {"name": "Auth", "description": "Authentication endpoints"},
    ],
    docs_url="/docs",        # Swagger UI
    redoc_url="/redoc",      # ReDoc
    openapi_url="/openapi.json",
)

Examples

示例

JWT Auth Dependency Chain

JWT认证依赖链

Complete auth chain from token to authorized user:

Request with Authorization: Bearer <token>
    ↓
oauth2_scheme (extracts token from header)
    ↓
get_current_user (decodes JWT, loads user from DB)
    ↓
require_role("admin") (checks user.role)
    ↓
Route handler (receives verified admin user)

Each dependency in the chain is independently testable via

dependency_overrides

从令牌到已授权用户的完整认证链：

Request with Authorization: Bearer <token>
    ↓
oauth2_scheme (extracts token from header)
    ↓
get_current_user (decodes JWT, loads user from DB)
    ↓
require_role("admin") (checks user.role)
    ↓
Route handler (receives verified admin user)

链中的每个依赖都可通过

dependency_overrides

独立测试。

Edge Cases

边缘情况

Middleware vs dependency: Use middleware for cross-cutting concerns (logging, timing, CORS). Use dependencies for per-route logic (auth, pagination params, feature flags).
ASGI vs BaseHTTPMiddleware: Prefer ASGI middleware for performance.
```
BaseHTTPMiddleware
```
reads the entire response body into memory, causing issues with streaming and large responses.
Lifespan vs on_event: Always use the
```
lifespan
```
context manager.
```
@app.on_event("startup")
```
and
```
@app.on_event("shutdown")
```
are deprecated in FastAPI 0.109+.
Depends caching across sub-applications: Dependency caching works per-request within a single app instance. If using
```
app.mount()
```
for sub-applications, each sub-app has its own dependency resolution scope.
WebSocket scaling: A single server instance holds all WebSocket connections. For multi-instance deployments, use Redis pub/sub to broadcast messages across instances.

See

references/middleware-examples.md

for complete middleware implementations. See

references/dependency-injection-patterns.md

for advanced DI patterns.

中间件 vs 依赖： 中间件用于横切关注点（日志、计时、CORS）。依赖用于路由级逻辑（认证、分页参数、功能开关）。
ASGI vs BaseHTTPMiddleware： 优先使用ASGI中间件以获得更好性能。
```
BaseHTTPMiddleware
```
会将整个响应体读入内存，可能导致流式响应和大响应的问题。
Lifespan vs on_event： 始终使用
```
lifespan
```
上下文管理器。
```
@app.on_event("startup")
```
和
```
@app.on_event("shutdown")
```
在FastAPI 0.109+中已被弃用。
子应用间的依赖缓存： 依赖缓存在单个应用实例的请求范围内生效。如果使用
```
app.mount()
```
挂载子应用，每个子应用都有独立的依赖解析范围。
WebSocket扩容： 单个服务器实例会持有所有WebSocket连接。对于多实例部署，请使用Redis发布/订阅在实例间广播消息。

查看

references/middleware-examples.md

获取完整的中间件实现。查看

references/dependency-injection-patterns.md

获取高级依赖注入模式。