---

name: fastapi-patterns description: FastAPI 模式：异步 API、依赖注入、Pydantic 请求和响应模型、OpenAPI 文档、测试、安全性和生产就绪。 origin: community

FastAPI 模式

面向生产的 FastAPI 服务模式。

何时使用

• 构建或审查 FastAPI 应用。
• 拆分路由器、模式、依赖项和数据库访问。
• 编写调用数据库或外部服务的异步端点。
• 添加认证、授权、OpenAPI 文档、测试或部署设置。
• 检查 FastAPI PR 中可复制粘贴的示例和生产风险。

工作原理

将 FastAPI 应用视为显式依赖和服务代码之上的薄 HTTP 层：

• main.py 负责应用构建、中间件、异常处理器和路由器注册。
• schemas/ 负责 Pydantic 请求和响应模型。
• dependencies.py 负责数据库、认证、分页和请求作用域依赖。
• services/ 或 crud/ 负责业务和持久化操作。
• tests/ 覆盖依赖项而非打开生产资源。

优先使用小型路由器和显式 response_model 声明。将原始 ORM 对象、密钥和框架全局变量排除在响应模式之外。

项目布局

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/

应用工厂

使用工厂模式，以便测试和工作进程可以使用受控设置构建应用。

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_origins=["*"] 与 allow_credentials=True 一起使用；浏览器会拒绝该组合，Starlette 也不允许凭证请求使用该组合。

Pydantic 模式

将请求、更新和响应模型分开。

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

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

依赖项

使用依赖注入管理请求作用域资源。

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

避免在路由处理器内部内联创建会话、客户端或凭证。

异步端点

当路由处理器执行 I/O 操作时保持异步。