fastapi-style

name	fastapi-style
description	FastAPI 框架编码风格 Skill。蒸馏自 tiangolo/fastapi 源码、Sebastián Ramírez 的设计哲学、 官方文档最佳实践、Pydantic v2 迁移指南以及 FastAPI 核心贡献者的 PR review 习惯。 触发词：「FastAPI 风格」「tiangolo 风格」「fastapi style」「Pydantic 最佳实践」「FastAPI API 设计」。 适用：FastAPI Web 服务开发、REST API 设计、Pydantic 模型建模、依赖注入系统、OpenAPI 文档生成。

FastAPI · 编码 DNA

"FastAPI is designed to be easy to use, but also to make it hard to make mistakes." — Sebastián Ramírez "Pydantic is not a validation library. It's a data modeling library that validates." — Samuel Colvin

角色定义

此 Skill 激活后，你写出的代码应该让 FastAPI 核心 contributor 在 code review 时感觉 「这代码把 Pydantic、依赖注入和 OpenAPI 三者用得浑然一体」， 而不是「这是 Flask 开发者用 FastAPI 写的 Flask」。

这意味着：类型注解驱动、依赖注入优先、Pydantic 模型精确建模、OpenAPI 文档是一等公民。

---

命名 DNA

5 条直觉规则：

1. Pydantic 模型名用 PascalCase，并体现领域语义（不叫 Data 或 Schema）

   # ✅ 领域驱动命名
   class UserCreate(BaseModel):     # 创建请求体
       email: EmailStr
       password: SecretStr
   
   class UserPublic(BaseModel):     # 响应体（对外暴露）
       id: int
       email: EmailStr
       created_at: datetime
   
   class UserUpdate(BaseModel):     # 更新请求体（全部可选）
       email: EmailStr | None = None
       display_name: str | None = None
   
   # ❌ 语义模糊
   class UserData(BaseModel): ...
   class UserSchema(BaseModel): ...  # Pydantic 本身就是 schema，后缀冗余
   class User(BaseModel): ...        # 和 ORM 模型同名，会冲突
   

2. 路由函数名用 动词_名词 snake_case，准确描述操作

   # ✅ 动词清晰表达 HTTP 语义
   @router.post("/users", response_model=UserPublic)
   async def create_user(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
       ...
   
   @router.get("/users/{user_id}", response_model=UserPublic)
   async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
       ...
   
   @router.patch("/users/{user_id}", response_model=UserPublic)
   async def update_user(user_id: int, user_in: UserUpdate, db: AsyncSession = Depends(get_db)):
       ...
   
   # ❌ 语义不清
   @router.post("/users")
   async def users(data: dict):  # data: dict 丢失类型信息
       ...
   
   @router.get("/users/{id}")
   async def user_handler(id: int):  # "handler" 不说明做什么
       ...
   

3. 依赖函数命名：get_ 前缀表示资源获取，require_ 前缀表示带校验的守卫

   # ✅ 语义一目了然
   async def get_db() -> AsyncGenerator[AsyncSession, None]:
       async with SessionLocal() as session:
           yield session
   
   async def get_current_user(
       token: str = Depends(oauth2_scheme),
       db: AsyncSession = Depends(get_db),
   ) -> User:
       ...
   
   async def require_admin(
       current_user: User = Depends(get_current_user),
   ) -> User:
       if not current_user.is_admin:
           raise HTTPException(status_code=403, detail="Admin required")
       return current_user
   
   # ❌ 命名模糊
   async def db_session(): ...       # 不知道是 get 还是 create
   async def check_user(): ...       # check 比 require 模糊
   

4. 路由路径用复数名词，路径参数用 snake_case

   # ✅ RESTful 约定
   @router.get("/users")                    # 集合
   @router.get("/users/{user_id}")          # 单个资源
   @router.get("/users/{user_id}/orders")  # 嵌套资源
   @router.post("/users/{user_id}/activate") # 动作
   
   # ❌
   @router.get("/user")           # 单数（不 RESTful）
   @router.get("/users/{userId}") # camelCase 路径参数（不 Pythonic）
   @router.get("/getUser")        # 动词路径（RPC 风格，不 REST）
   

5. settings/config 用 pydantic-settings 的 BaseSettings，不用裸 os.environ

   # ✅ 类型安全的配置，自动从环境变量读取
   from pydantic_settings import BaseSettings, SettingsConfigDict
   
   class Settings(BaseSettings):
       model_config = SettingsConfigDict(env_file=".env", env_ignore_empty=True)
   
       database_url: str
       secret_key: SecretStr
       debug: bool = False
       max_connections: int = 10
   
   settings = Settings()
   
   # ❌ 散落各处的 os.getenv
   DATABASE_URL = os.getenv("DATABASE_URL")
   SECRET_KEY = os.environ["SECRET_KEY"]  # 类型是 str，但可能是密钥，应该用 SecretStr
   

---

结构偏好

项目目录结构（中型项目标准布局）：

app/
├── main.py              # FastAPI app 创建、router 注册、lifespan
├── config.py            # Settings（pydantic-settings）
├── dependencies.py      # 全局共用依赖（get_db、get_current_user）
├── api/
│   ├── v1/
│   │   ├── router.py    # v1 总路由（include_router 汇总）
│   │   ├── users.py     # users 资源路由
│   │   └── orders.py    # orders 资源路由
├── models/
│   ├── user.py          # ORM 模型（SQLAlchemy）
│   └── order.py
├── schemas/             # 或叫 dto/ — Pydantic 模型
│   ├── user.py          # UserCreate, UserPublic, UserUpdate
│   └── order.py
├── services/            # 业务逻辑层（不含路由、不含 DB 细节）
│   ├── user_service.py
│   └── order_service.py
└── db/
    ├── session.py       # 数据库 session 工厂
    └── base.py          # ORM Base

Pydantic v2 模型设计原则：

# ✅ 用 model_config 替代 class Config
from pydantic import BaseModel, ConfigDict, Field, EmailStr

class UserPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # v2 风格（替代 orm_mode=True）

    id: int
    email: EmailStr
    display_name: str = Field(min_length=1, max_length=100)
    created_at: datetime

    # ✅ 敏感字段用 exclude 或单独定义响应模型
    # 不在 UserPublic 里暴露 hashed_password

# ✅ 请求体验证逻辑用 @field_validator（v2 风格）
class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8)
    confirm_password: str

    @model_validator(mode='after')
    def passwords_match(self) -> 'UserCreate':
        if self.password != self.confirm_password:
            raise ValueError('passwords do not match')
        return self

# ❌ v1 过时写法（Pydantic v2 已弃用）
class OldUser(BaseModel):
    class Config:
        orm_mode = True   # v2 用 from_attributes=True
    
    @validator('email')   # v2 用 @field_validator
    def validate_email(cls, v): ...

依赖注入分层：路由层薄，服务层厚

# ✅ 路由函数只做：参数接收 → 调用服务 → 返回响应
@router.post("/users", response_model=UserPublic, status_code=201)
async def create_user(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db),
    # 注意：不把 UserService 实例化放这里
) -> UserPublic:
    user = await UserService(db).create(user_in)
    return user

# ✅ 服务层包含业务逻辑
class UserService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def create(self, user_in: UserCreate) -> User:
        if await self._email_exists(user_in.email):
            raise HTTPException(status_code=409, detail="Email already registered")
        hashed = hash_password(user_in.password)
        user = User(email=user_in.email, hashed_password=hashed)
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def _email_exists(self, email: str) -> bool:
        result = await self.db.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none() is not None

# ❌ 路由函数里堆业务逻辑和 DB 操作
@router.post("/users")
async def create_user(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
    # 50行业务逻辑全在路由函数里...
    existing = await db.execute(select(User).where(User.email == user_in.email))
    if existing.scalar_one_or_none():
        raise HTTPException(...)
    # ...

lifespan 管理应用生命周期（替代 on_event）：

# ✅ FastAPI 0.93+ 用 lifespan 上下文管理器
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    await db_pool.connect()
    await redis_client.ping()
    logger.info("application started")

    yield  # 应用运行中

    # shutdown
    await db_pool.disconnect()
    await redis_client.aclose()
    logger.info("application stopped")

app = FastAPI(lifespan=lifespan)

# ❌ 已弃用的 on_event 写法
@app.on_event("startup")
async def startup(): ...
@app.on_event("shutdown")
async def shutdown(): ...

OpenAPI 文档是一等公民：每个路由都要有完整注释

# ✅ 完整的路由文档
@router.post(
    "/users",
    response_model=UserPublic,
    status_code=201,
    summary="创建新用户",
    description="注册新用户账号，邮箱需唯一。密码会自动 bcrypt 哈希存储，不会明文保存。",
    responses={
        201: {"description": "用户创建成功"},
        409: {"description": "邮箱已被注册", "model": ErrorResponse},
        422: {"description": "请求体格式错误"},
    },
    tags=["users"],
)
async def create_user(
    user_in: UserCreate = Body(..., examples={
        "standard": {
            "summary": "普通注册",
            "value": {"email": "[e-mail]", "password": "securepass123"},
        }
    }),
    db: AsyncSession = Depends(get_db),
) -> UserPublic:
    ...

# ❌ 裸路由，OpenAPI 文档无意义
@router.post("/users")
async def create_user(data: dict, db=Depends(get_db)):
    ...

---

注释哲学

docstring 面向使用者，解释业务语义而不是代码逻辑：

class UserService:
    """用户领域服务。

    负责用户账号的创建、查询、更新和鉴权逻辑。
    所有数据库操作通过注入的 AsyncSession 进行，
    不直接管理事务（由调用方或中间件管理）。
    """

    async def create(self, user_in: UserCreate) -> User:
        """创建新用户账号。

        Args:
            user_in: 用户注册信息，包含 email 和明文密码。

        Returns:
            创建成功的 User ORM 对象（已 refresh，包含 DB 生成的 id 和 created_at）。

        Raises:
            HTTPException(409): 邮箱已被注册。
            HTTPException(500): 数据库写入失败。
        """

类型注解优先于 docstring 解释参数类型：

# ✅ 类型注解已经说明了类型，docstring 说明语义
async def get_user_by_email(email: EmailStr, db: AsyncSession) -> User | None:
    """通过邮箱查找用户，不存在时返回 None（不抛异常）。"""

# ❌ docstring 重复类型信息（类型注解已经说了）
async def get_user_by_email(email: str, db) -> User:
    """
    Args:
        email (str): 用户邮箱字符串
        db (AsyncSession): 数据库会话对象
    Returns:
        User: 用户对象
    """

---

反模式（绝不这样写）

1. 路由函数接收 dict 或 Any（丢失类型校验和 OpenAPI 文档）

   # ❌ dict 无法自动生成 OpenAPI schema，也无法校验
   @router.post("/users")
   async def create_user(body: dict): ...
   
   # ✅ 用 Pydantic 模型
   @router.post("/users")
   async def create_user(user_in: UserCreate): ...
   

2. 在 Pydantic 模型里暴露 ORM 内部字段（信息泄露）

   # ❌ 把 ORM 模型直接作为响应体 — 可能泄露 hashed_password、内部 id 等
   @router.get("/users/{user_id}")
   async def get_user(user_id: int, db: AsyncSession = Depends(get_db)) -> User:  # ORM 模型！
       return await db.get(User, user_id)
   
   # ✅ 明确的响应 schema
   @router.get("/users/{user_id}", response_model=UserPublic)
   async def get_user(user_id: int, db: AsyncSession = Depends(get_db)) -> UserPublic:
       user = await db.get(User, user_id)
       if not user:
           raise HTTPException(status_code=404, detail="User not found")
       return user  # FastAPI 会用 UserPublic 序列化
   

3. 把数据库会话作为全局变量（多并发下线程/协程安全问题）

   # ❌ 全局 session，并发请求共享状态 = 数据竞争
   db = SessionLocal()
   
   @router.get("/users")
   async def list_users():
       return db.query(User).all()
   
   # ✅ 每个请求独立 session，通过依赖注入
   @router.get("/users")
   async def list_users(db: AsyncSession = Depends(get_db)):
       result = await db.execute(select(User))
       return result.scalars().all()
   

4. HTTPException 的 detail 暴露内部错误信息（安全问题）

   # ❌ 把 DB 错误或堆栈信息暴露给客户端
   try:
       await db.commit()
   except Exception as e:
       raise HTTPException(status_code=500, detail=str(e))  # 暴露内部实现！
   
   # ✅ 日志记录内部错误，对外返回通用消息
   try:
       await db.commit()
   except Exception:
       logger.exception("database commit failed")
       raise HTTPException(status_code=500, detail="Internal server error")
   

5. response_model 和函数返回类型注解不一致

   # ❌ 类型注解骗人（函数签名说返回 User，但 response_model 是 UserPublic）
   @router.get("/users/{user_id}", response_model=UserPublic)
   async def get_user(user_id: int) -> User:  # 实际序列化用 UserPublic
       ...
   
   # ✅ 统一
   @router.get("/users/{user_id}", response_model=UserPublic)
   async def get_user(user_id: int) -> UserPublic:
       ...
   

6. 同步阻塞操作在 async 路由里直接调用（阻塞事件循环）

   # ❌ 同步 requests 库在 async 函数里 — 阻塞整个事件循环
   @router.get("/proxy")
   async def proxy():
       response = requests.get("https://api.example.com/data")  # 同步阻塞！
       return response.json()
   
   # ✅ 用 httpx 的 async 客户端
   @router.get("/proxy")
   async def proxy(client: httpx.AsyncClient = Depends(get_http_client)):
       response = await client.get("https://api.example.com/data")
       return response.json()
   

7. 依赖链里有副作用但没有用 yield（资源泄漏）

   # ❌ 没有用 yield，close 不保证执行
   async def get_db() -> AsyncSession:
       db = SessionLocal()
       return db  # 如果路由抛异常，db 永远不会 close！
   
   # ✅ yield 依赖保证 finally 执行
   async def get_db() -> AsyncGenerator[AsyncSession, None]:
       async with SessionLocal() as db:
           try:
               yield db
               await db.commit()
           except Exception:
               await db.rollback()
               raise
   

---

校验测试

1. 类型覆盖率：路由函数的所有参数（路径参数、Query、Body、Depends）有没有显式类型注解？dict 和 Any 应该是零。
2. OpenAPI 完整性：在 /docs 页面看每个路由，有没有 summary、正确的 response_model、错误码的 responses？能不能直接在 Swagger UI 里测试？
3. 依赖隔离：路由函数是不是「薄」的（<10行）？业务逻辑有没有下沉到服务层或依赖函数里？

---

来源

• tiangolo/fastapi 源码和官方文档
• FastAPI Best Practices — 社区最佳实践汇总
• Sebastián Ramírez 的 FastAPI 设计哲学：FastAPI People
• pydantic/pydantic v2 迁移指南
• pydantic/pydantic-settings 配置管理
• SQLAlchemy 2.0 异步文档：AsyncSession