Back to Details

fastapi-patterns

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

FastAPI Patterns

FastAPI 模式

Production-oriented patterns for FastAPI services.
面向生产环境的FastAPI服务模式。

When to Use

使用场景

  • Building or reviewing a FastAPI app.
  • Splitting routers, schemas, dependencies, and database access.
  • Writing async endpoints that call a database or external service.
  • Adding authentication, authorization, OpenAPI docs, tests, or deployment settings.
  • Checking a FastAPI PR for copy-pasteable examples and production risks.
  • 构建或评审FastAPI应用。
  • 拆分路由、Schema、依赖项和数据库访问逻辑。
  • 编写调用数据库或外部服务的异步端点。
  • 添加认证、授权、OpenAPI文档、测试或部署配置。
  • 检查FastAPI的PR，获取可复制的示例并排查生产风险。

How It Works

工作原理

Treat the FastAPI app as a thin HTTP layer over explicit dependencies and service code:
  • main.py
    owns app construction, middleware, exception handlers, and router registration.
  • schemas/
    owns Pydantic request and response models.
  • dependencies.py
    owns database, auth, pagination, and request-scoped dependencies.
  • services/
    or 
    crud/
    owns business and persistence operations.
  • tests/
    overrides dependencies instead of opening production resources.
Prefer small routers and explicit 
response_model
declarations. Keep raw ORM objects, secrets, and framework globals out of response schemas.
将FastAPI应用视为显式依赖项与业务代码之上的轻量HTTP层：
  • main.py
    负责应用构建、中间件、异常处理器和路由注册。
  • schemas/
    目录存放Pydantic请求与响应模型。
  • dependencies.py
    负责数据库、认证、分页和请求作用域的依赖项。
  • services/
    或
    crud/
    目录存放业务逻辑与持久化操作。
  • tests/
    通过覆盖依赖项而非直接访问生产资源来实现测试。
优先使用小型路由和显式的
response_model
声明。避免在响应Schema中暴露原始ORM对象、密钥和框架全局变量。

Project Layout

项目结构

text
app/
|-- main.py
|-- config.py
|-- dependencies.py
|-- exceptions.py
|-- api/
|   `-- routes/
|       |-- users.py
|       `-- health.py
|-- core/
|   |-- security.py
|   `-- middleware.py
|-- db/
|   |-- session.py
|   `-- crud.py
|-- models/
|-- schemas/
`-- tests/
text
app/
|-- main.py
|-- config.py
|-- dependencies.py
|-- exceptions.py
|-- api/
|   `-- routes/
|       |-- users.py
|       `-- health.py
|-- core/
|   |-- security.py
|   `-- middleware.py
|-- db/
|   |-- session.py
|   `-- crud.py
|-- models/
|-- schemas/
`-- tests/

Application Factory

应用工厂

Use a factory so tests and workers can build the app with controlled settings.
python
from contextlib import asynccontextmanager

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

from app.api.routes import health, users
from app.config import settings
from app.db.session import close_db, init_db
from app.exceptions import register_exception_handlers


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield
    await close_db()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.api_title,
        version=settings.api_version,
        lifespan=lifespan,
    )

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=bool(settings.cors_origins),
        allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
        allow_headers=["Authorization", "Content-Type"],
    )

    register_exception_handlers(app)
    app.include_router(health.router, prefix="/health", tags=["health"])
    app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
    return app


app = create_app()
Do not use 
allow_origins=["*"]
with 
allow_credentials=True
; browsers reject that combination and Starlette disallows it for credentialed requests.
使用工厂模式，以便测试和工作进程可以通过受控配置构建应用。
python
from contextlib import asynccontextmanager

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

from app.api.routes import health, users
from app.config import settings
from app.db.session import close_db, init_db
from app.exceptions import register_exception_handlers


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield
    await close_db()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.api_title,
        version=settings.api_version,
        lifespan=lifespan,
    )

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=bool(settings.cors_origins),
        allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
        allow_headers=["Authorization", "Content-Type"],
    )

    register_exception_handlers(app)
    app.include_router(health.router, prefix="/health", tags=["health"])
    app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
    return app


app = create_app()
请勿在
allow_credentials=True
时使用
allow_origins=["*"]
；浏览器会拒绝这种组合，Starlette也不允许针对带凭证的请求使用该配置。

Pydantic Schemas

Pydantic Schema

Keep request, update, and response models separate.
python
from datetime import datetime
from typing import Annotated
from uuid import UUID

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserBase(BaseModel):
    email: EmailStr
    full_name: Annotated[str, Field(min_length=1, max_length=100)]


class UserCreate(UserBase):
    password: Annotated[str, Field(min_length=12, max_length=128)]


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    full_name: Annotated[str | None, Field(min_length=1, max_length=100)] = None


class UserResponse(UserBase):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    created_at: datetime
    updated_at: datetime
Response models must never include password hashes, access tokens, refresh tokens, or internal authorization state.
将请求、更新和响应模型分开存放。
python
from datetime import datetime
from typing import Annotated
from uuid import UUID

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserBase(BaseModel):
    email: EmailStr
    full_name: Annotated[str, Field(min_length=1, max_length=100)]


class UserCreate(UserBase):
    password: Annotated[str, Field(min_length=12, max_length=128)]


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    full_name: Annotated[str | None, Field(min_length=1, max_length=100)] = None


class UserResponse(UserBase):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    created_at: datetime
    updated_at: datetime
响应模型绝不能包含密码哈希、访问令牌、刷新令牌或内部授权状态。

Dependencies

依赖项

Use dependency injection for request-scoped resources.
python
from collections.abc import AsyncIterator
from uuid import UUID

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import decode_token
from app.db.session import session_factory
from app.models.user import User


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


async def get_db() -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = decode_token(token)
    user_id = UUID(payload["sub"])
    user = await db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return user
Avoid creating sessions, clients, or credentials inline inside route handlers.
使用依赖注入管理请求作用域的资源。
python
from collections.abc import AsyncIterator
from uuid import UUID

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import decode_token
from app.db.session import session_factory
from app.models.user import User


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


async def get_db() -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = decode_token(token)
    user_id = UUID(payload["sub"])
    user = await db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return user
避免在路由处理器内联创建会话、客户端或凭证。

Async Endpoints

异步端点

Keep route handlers async when they perform I/O, and use async libraries inside them.
python
from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_current_user, get_db
from app.models.user import User
from app.schemas.user import UserResponse


router = APIRouter()


@router.get("/", response_model=list[UserResponse])
async def list_users(
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    result = await db.execute(
        select(User).order_by(User.created_at.desc()).limit(limit).offset(offset)
    )
    return result.scalars().all()
Use 
httpx.AsyncClient
for external HTTP calls from async handlers. Do not call 
requests
in an async route.
当端点执行I/O操作时，保持路由处理器为异步模式，并在其中使用异步库。
python
from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_current_user, get_db
from app.models.user import User
from app.schemas.user import UserResponse


router = APIRouter()


@router.get("/", response_model=list[UserResponse])
async def list_users(
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    result = await db.execute(
        select(User).order_by(User.created_at.desc()).limit(limit).offset(offset)
    )
    return result.scalars().all()
在异步处理器中使用
httpx.AsyncClient
发起外部HTTP请求。请勿在异步路由中调用
requests
库。

Error Handling

错误处理

Centralize domain exceptions and keep response shapes stable.
python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class ApiError(Exception):
    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code = code
        self.message = message


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(ApiError)
    async def api_error_handler(request: Request, exc: ApiError):
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": {"code": exc.code, "message": exc.message}},
        )
集中管理领域异常，保持响应格式稳定。
python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class ApiError(Exception):
    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code = code
        self.message = message


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(ApiError)
    async def api_error_handler(request: Request, exc: ApiError):
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": {"code": exc.code, "message": exc.message}},
        )

OpenAPI Customization

OpenAPI 自定义

Assign the custom OpenAPI callable to 
app.openapi
; do not just call the function once.
python
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi


def install_openapi(app: FastAPI) -> None:
    def custom_openapi():
        if app.openapi_schema:
            return app.openapi_schema
        app.openapi_schema = get_openapi(
            title="Service API",
            version="1.0.0",
            routes=app.routes,
        )
        return app.openapi_schema

    app.openapi = custom_openapi
将自定义OpenAPI可调用对象赋值给
app.openapi
；请勿仅调用该函数一次。
python
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi


def install_openapi(app: FastAPI) -> None:
    def custom_openapi():
        if app.openapi_schema:
            return app.openapi_schema
        app.openapi_schema = get_openapi(
            title="Service API",
            version="1.0.0",
            routes=app.routes,
        )
        return app.openapi_schema

    app.openapi = custom_openapi

Testing

测试

Override the dependency used by 
Depends
, not an internal helper that route handlers never reference.
python
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db
from app.main import create_app


@pytest.fixture
async def client(test_session: AsyncSession):
    app = create_app()

    async def override_get_db():
        yield test_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as test_client:
        yield test_client
    app.dependency_overrides.clear()
覆盖
Depends
使用的依赖项，而非路由处理器从未引用的内部辅助函数。
python
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db
from app.main import create_app


@pytest.fixture
async def client(test_session: AsyncSession):
    app = create_app()

    async def override_get_db():
        yield test_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as test_client:
        yield test_client
    app.dependency_overrides.clear()

Security Checklist

安全检查清单

  • Hash passwords with 
    argon2-cffi
    , 
    bcrypt
    , or a current passlib-compatible hasher.
  • Validate JWT issuer, audience, expiry, and signing algorithm.
  • Keep CORS origins environment-specific.
  • Put rate limits on auth and write-heavy endpoints.
  • Use Pydantic models for all request bodies.
  • Use ORM parameter binding or SQLAlchemy Core expressions; never build SQL with f-strings.
  • Redact tokens, authorization headers, cookies, and passwords from logs.
  • Run dependency audit tooling in CI.
  • 使用
    argon2-cffi
    、
    bcrypt
    或兼容passlib的当前哈希算法对密码进行哈希处理。
  • 验证JWT的签发者、受众、过期时间和签名算法。
  • 保持CORS源与环境相关。
  • 对认证和写密集型端点设置速率限制。
  • 对所有请求体使用Pydantic模型。
  • 使用ORM参数绑定或SQLAlchemy Core表达式；绝不要使用f-string构建SQL语句。
  • 在日志中屏蔽令牌、授权头、Cookie和密码。
  • 在CI中运行依赖项审计工具。

Performance Checklist

性能检查清单

  • Configure database connection pooling explicitly.
  • Add pagination to list endpoints.
  • Watch for N+1 queries and use eager loading intentionally.
  • Use async HTTP/database clients in async paths.
  • Add compression only after checking payload size and CPU tradeoffs.
  • Cache stable expensive reads behind explicit invalidation.
  • 显式配置数据库连接池。
  • 为列表端点添加分页功能。
  • 注意N+1查询问题，有意使用预加载。
  • 在异步路径中使用异步HTTP/数据库客户端。
  • 仅在检查有效负载大小和CPU权衡后添加压缩。
  • 在显式失效策略后缓存稳定的高开销读取操作。

Examples

示例

Use these examples as patterns, not as project-wide templates:
  • Application factory: configure middleware and routers once in 
    create_app
    .
  • Schema split: 
    UserCreate
    , 
    UserUpdate
    , and 
    UserResponse
    have different responsibilities.
  • Dependency override: tests override 
    get_db
    directly.
  • OpenAPI customization: assign 
    app.openapi = custom_openapi
    .
将这些示例作为模式使用，而非全项目模板：
  • 应用工厂：在
    create_app
    中一次性配置中间件和路由。
  • Schema拆分：
    UserCreate
    、
    UserUpdate
    和
    UserResponse
    具有不同职责。
  • 依赖项覆盖：测试直接覆盖
    get_db
    。
  • OpenAPI自定义：赋值
    app.openapi = custom_openapi
    。

See Also

另请参阅

  • Agent: 
    fastapi-reviewer
  • Command: 
    /fastapi-review
  • Skill: 
    python-patterns
  • Skill: 
    python-testing
  • Skill: 
    api-design
  • Agent: 
    fastapi-reviewer
  • Command: 
    /fastapi-review
  • Skill: 
    python-patterns
  • Skill: 
    python-testing
  • Skill: 
    api-design