一文读懂FastAPI 最佳实践指南 (2025 生产环境版)

目录

1. 项目结构

2. 异步路由与并发模型

3. Pydantic 深度应用 (V2)

4. 依赖项 (Dependencies) 的艺术

5. 数据库与 ORM (SQLAlchemy 2.0)

6. 安全性与生产级配置

7. 文档与规范

8. 测试策略

9. 部署与可观测性

10. 工具链 (Ruff)

项目结构

项目结构有很多种，但最好的结构是一致、直观且没有意外的。

许多教程按文件类型（如 crud, routers, models）划分，这对微服务有效，但对单体应用不友好。我们推荐基于 领域模块 (Domain-based) 的结构（灵感来自 Netflix Dispatch），并根据现代 Python 习惯进行了微调。

fastapi-project
├── alembic/
├── src
│   ├── auth/              # 领域模块：认证
│   │   ├── router.py
│   │   ├── schemas.py     # Pydantic 模型
│   │   ├── models.py      # 数据库模型
│   │   ├── dependencies.py
│   │   ├── config.py      # 模块级配置
│   │   ├── service.py     # 业务逻辑
│   │   └── exceptions.py
│   ├── posts/             # 领域模块：帖子
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   ├── service.py
│   │   └── ...
│   ├── config.py          # 全局配置
│   ├── database.py        # 数据库连接与 Session 管理
│   ├── main.py            # 应用入口
│   ├── pagination.py      # 通用工具
│   └── utils.py
├── tests/
├── templates/
├── requirements/
│   ├── base.txt
│   └── prod.txt
├── .env
├── logging.ini (或使用 structlog 配置)
└── alembic.ini

核心原则：

• src/: 应用的根目录。

• main.py: 初始化 FastAPI 应用。

• 领域隔离: 每个包（如 auth, posts）拥有自己的路由、模型和服务。

• 显式导入: 当模块间需要交互时，使用完整路径导入，例如 from src.auth import service as auth_service。

异步路由与并发模型

FastAPI 的核心优势是异步 I/O，但必须正确理解 async 和 def 的区别。

I/O 密集型任务

FastAPI 处理机制：

1. 同步路由 (def): 在线程池中运行。阻塞操作不会卡死主事件循环。

2. 异步路由 (async def): 在主事件循环中运行。严禁执行阻塞操作。

反面教材与正确示例：

import asyncio
import time
from fastapi import APIRouter

router = APIRouter()

@router.get("/terrible-ping")
async def terrible_ping():
    time.sleep(10) # 灾难！阻塞整个进程，所有请求都会被卡住
    return {"pong": True}

@router.get("/good-ping")
def good_ping():
    time.sleep(10) # 安全。在线程池运行，只阻塞该线程
    return {"pong": True}

@router.get("/perfect-ping")
async def perfect_ping():
    await asyncio.sleep(10) # 完美。非阻塞，释放控制权给其他请求
    return {"pong": True}

CPU 密集型任务

不要在 FastAPI 主进程或线程池中运行繁重的 CPU 任务（如视频转码、复杂计算），因为 GIL（全局解释器锁）会限制并发。

• 解决方案: 使用 Celery、ARQ 或其他 Worker 进程处理。

同步 SDK 的处理

如果必须使用同步库（且无法重构为异步），请显式放入线程池：

from fastapi.concurrency import run_in_threadpool

@app.get("/")
async def call_sync_lib():
    data = await service.get_data()
    # 在线程池中执行同步请求
    result = await run_in_threadpool(sync_client.request, data)
    return result

Pydantic 深度应用 (V2)

Pydantic 是数据验证和序列化的核心。

1. 使用 computed_field (V2 新特性)

替代复杂的 getter 方法，性能更好且自动序列化。

from pydantic import BaseModel, computed_field

class UserProfile(BaseModel):
    first_name: str
    last_name: str

    @computed_field
    @property
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

2. 自定义基础模型与序列化

控制全局配置，例如时间格式。注意：在 V2 中，优先使用 model_dump(mode='json')。

from datetime import datetime
from pydantic import BaseModel, ConfigDict

class CustomModel(BaseModel):
    model_config = ConfigDict(
        populate_by_name=True,
        str_strip_whitespace=True, # 自动去除字符串首尾空格
    )

    def serializable_dict(self, **kwargs):
        return self.model_dump(mode="json", exclude_unset=True)

3. 拆分 Pydantic BaseSettings

不要把所有环境变量放在一个文件里，按模块拆分。

# src/auth/config.py
from pydantic_settings import BaseSettings

class AuthConfig(BaseSettings):
    JWT_SECRET: str
    JWT_EXP_MINUTES: int = 60

auth_settings = AuthConfig()

4. ValueError 与 ValidationError

直接在 Validator 中抛出 ValueError，FastAPI 会自动将其转换为 422 Unprocessable Entity 响应，包含详细错误信息。

依赖项 (Dependencies) 的艺术

依赖项不仅用于注入，更是强大的请求验证层。

1. 验证而非仅注入

将业务前置检查（如“帖子是否存在”、“用户是否有权”）放入依赖项。

async def valid_post_id(post_id: UUID4) -> dict:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()
    return post

@router.get("/posts/{post_id}")
async def get_post(post: dict = Depends(valid_post_id)):
    return post

2. 链式依赖与缓存

FastAPI 默认缓存依赖项结果。如果一个请求中多次调用 valid_post_id，它只执行一次。

async def valid_owned_post(
    post: dict = Depends(valid_post_id), 
    token_data: dict = Depends(parse_jwt_data),
) -> dict:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()
    return post

3. 使用 yield 管理资源生命周期

这是管理数据库 Session 最重要的方式。

# src/database.py
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        try:
            yield session
            # 请求成功处理后，可在此处 commit (如果配置了autocommit=False)
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

4. 基于类的依赖项

当依赖参数过多时，使用类来组织。

class PaginationDep:
    def __init__(self, page: int = 1, size: int = 20):
        self.skip = (page - 1) * size
        self.limit = size

@router.get("/items")
async def list_items(page: PaginationDep = Depends()):
    return service.get(skip=page.skip, limit=page.limit)

5. 优先使用 Async 依赖

尽量避免同步依赖项，因为它们会在线程池中运行，增加开销。

数据库与 ORM (SQLAlchemy 2.0)

1. SQL 优先，Pydantic 次之

Python 处理数据慢，数据库快。尽量在数据库层面完成 Join、聚合和过滤。拥抱 SQLAlchemy 2.0 语法。

# 推荐的 2.0 风格写法
async def get_posts_with_creators(session: AsyncSession):
    stmt = (
        select(Post)
        .options(selectinload(Post.creator)) # 预加载避免 N+1
        .where(Post.public == True)
        .order_by(Post.created_at.desc())
    )
    result = await session.execute(stmt)
    return result.scalars().all()

2. 设置数据库键命名约定

显式配置 SQLAlchemy 的命名约定，避免迁移时的名字冲突。

metadata = MetaData(naming_convention={
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "pk": "pk_%(table_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
})

3. Alembic 迁移规范

• 静态且可回滚。

• 使用描述性 Slug：2025-01-01_add_user_index.py。

• 不要在迁移脚本中引用应用代码中的 Model 类，应直接写 Table 定义，防止未来代码变更破坏旧迁移。

安全性与生产级配置

1. 严格配置 CORS

在生产环境，永远不要使用 "*"。

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"], # 显式白名单
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

2. 速率限制 (Rate Limiting)

引入 slowapi 防止接口滥用。

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.get("/sensitive-action")
@limiter.limit("5/minute")
async def action(request: Request):
    ...

文档与规范

1. 环境敏感的文档

仅在开发/Staging 环境暴露 Swagger UI。

docs_url = "/docs" if settings.ENVIRONMENT != "production" else None
app = FastAPI(docs_url=docs_url)

2. 丰富的 OpenAPI 文档

利用 response_model, summary, description 和 responses 装饰器参数，让文档成为真正的说明书。

@router.post(
    "/posts", 
    status_code=201, 
    responses={409: {"description": "Post already exists"}}
)

测试策略

1. 异步测试客户端

必须使用 httpx 或 AsyncClient 进行集成测试。

import pytest
from httpx import AsyncClient, ASGITransport

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

@pytest.mark.asyncio
async def test_create_post(client):
    resp = await client.post("/posts", json={"title": "Test"})
    assert resp.status_code == 201

2. 使用 Factory Boy / Polyfactory

拒绝手动拼凑测试字典，使用工厂模式生成测试数据。

from polyfactory.factories.pydantic_factory import ModelFactory

class PostFactory(ModelFactory[PostCreate]):
    __model__ = PostCreate

def test_model(client):
    data = PostFactory.build()
    client.post("/posts", json=data.model_dump())

部署与可观测性

1. 结构化日志 (Structured Logging)

在生产环境中，文本日志难以检索。使用 structlog 输出 JSON 格式日志。

import structlog
logger = structlog.get_logger()

# 输出: {"event": "user_login", "user_id": 123, "level": "info", "timestamp": "..."}
await logger.info("user_login", user_id=123)

2. Docker 多阶段构建

保持镜像精简。

# Builder Stage
FROM python:3.11-slim as builder
COPY requirements.txt .
RUN pip install --user -r requirements.txt

# Final Stage
FROM python:3.11-slim
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH
COPY . /app
WORKDIR /app
CMD ["gunicorn", "src.main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000"]

3. Gunicorn + Uvicorn

在生产环境中，使用 Gunicorn 作为进程管理器，Uvicorn 作为 Worker 类。Worker 数量通常建议为 2 * CPU核心 + 1。

工具链 (Ruff)

使用 Ruff 替代 Black, Isort, Flake8。它速度极快且配置集中。

# pre-commit 脚本或 CI 流程
ruff check --fix src
ruff format src