Fastapi开发规范

FastAPI开发规范

路由、Pydantic 模型、依赖注入、中间件

项目结构

规则

[Requirement] 采用分层目录结构，按职责划分包：api（路由层）、schemas（Pydantic 模型）、services（业务逻辑）、repositories（数据访问）、core（配置/异常/安全）、models（ORM 模型）。
[Requirement] 入口文件 app/main.py 只负责创建 FastAPI 实例、注册路由与中间件，不承载业务逻辑。
[Standard] 路由按资源域拆分，每个域一个模块或包（app/api/v1/users.py），通过 APIRouter 聚合，在 main.py 中统一 include_router。
[Standard] 配置集中在 app/core/config.py，使用 pydantic-settings 从环境变量读取，禁止在代码中硬编码配置项。
[Standard] 常量（错误码、状态枚举）集中在 app/core/constants.py，禁止散落魔法值。
[Suggestion] 大型项目按 DDD 思路拆分子包 app/domains/user/，每个域内自带 api/schemas/services/repositories。

代码示例

myapp/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 入口
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── router.py       # v1 总路由聚合
│   │       └── users.py        # 用户相关路由
│   ├── schemas/                # Pydantic 模型
│   │   ├── user.py
│   │   └── common.py
│   ├── services/               # 业务逻辑
│   │   └── user_service.py
│   ├── repositories/           # 数据访问
│   │   └── user_repo.py
│   ├── models/                 # ORM 模型
│   │   └── user.py
│   ├── core/
│   │   ├── config.py           # 配置
│   │   ├── exceptions.py       # 自定义异常
│   │   └── security.py         # 鉴权
│   └── db/
│       └── session.py          # 数据库会话
├── tests/
├── pyproject.toml
└── .env.example

# app/main.py
from fastapi import FastAPI
 
from app.api.v1.router import api_router
from app.core.config import settings
 
app = FastAPI(title=settings.app_name, version="1.0.0")
app.include_router(api_router, prefix="/api/v1")

反模式

# 反模式：main.py 堆砌业务逻辑与数据库操作
@app.get("/users")
def get_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return [{"id": u.id, "name": u.name} for u in users]  # 逻辑全堆在路由

路由设计

规则

[Requirement] URL 使用名词复数表示资源（/users、/orders），禁止在 URL 中出现动词（/getUsers）。
[Requirement] HTTP 方法语义正确：GET 查询、POST 创建、PUT 整体更新、PATCH 部分更新、DELETE 删除。
[Requirement] 所有路由必须标注 response_model，利用 Pydantic 自动过滤输出字段，避免敏感字段泄露。
[Standard] 路由函数只做参数解析与响应组装，业务逻辑下沉到 service 层；路由函数体不超过 20 行。
[Standard] 路径参数做类型约束（user_id: int），查询参数用 Pydantic Query 限制范围，避免无效请求透传到服务层。
[Default] 统一响应结构：成功返回业务数据，失败由全局异常处理器返回统一错误体；禁止在路由内手动拼 {"code": 0, "data": ...}。
[Suggestion] 为每个路由添加 summary 与 description，生成清晰的 OpenAPI 文档。

代码示例

# app/api/v1/users.py
from fastapi import APIRouter, Depends, HTTPException, status
 
from app.schemas.user import UserCreate, UserOut
from app.services.user_service import UserService
 
router = APIRouter(prefix="/users", tags=["users"])
 
 
@router.get(
    "/{user_id}",
    response_model=UserOut,
    summary="获取用户详情",
)
def get_user(
    user_id: int,
    service: UserService = Depends(),
) -> UserOut:
    user = service.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user
 
 
@router.post(
    "",
    response_model=UserOut,
    status_code=status.HTTP_201_CREATED,
    summary="创建用户",
)
def create_user(payload: UserCreate, service: UserService = Depends()) -> UserOut:
    return service.create(payload)

# app/schemas/common.py — 统一响应结构
from typing import Any, Generic, TypeVar
from pydantic import BaseModel
 
T = TypeVar("T")
 
class ApiResponse(BaseModel, Generic[T]):
    code: int = 0
    message: str = "success"
    data: T | None = None

反模式

# 反模式：URL 含动词、方法语义错误
@router.get("/getUserById/{id}")       # 动词 + 路径参数未约束类型
def get_user(id, db):                   # 无类型注解
    user = db.query(User).get(id)
    return {"code": 0, "data": user}    # 手动拼响应体，泄露密码字段
 
# 反模式：路由内直接写 SQL 与业务逻辑
@router.post("/createUser")
def create_user(name: str, email: str, db):
    if not name or "@" not in email:
        return {"error": "invalid"}     # 非统一错误结构
    db.execute(f"INSERT INTO users ...") # SQL 注入风险
    db.commit()

Pydantic 模型

规则

[Requirement] 所有请求体与响应体必须定义 Pydantic 模型，禁止直接用 dict 接收或返回。
[Requirement] 请求模型与响应模型分离：创建用 XxxCreate、更新用 XxxUpdate、响应用 XxxOut，避免复用导致字段语义混乱。
[Standard] 使用 Pydantic v2 的 Field 对字段做约束（长度、范围、正则），把校验前置到模型层。
[Standard] 敏感字段（密码、token）仅出现在请求模型中，响应模型必须用 exclude 或不声明该字段来过滤。
[Standard] 枚举字段使用 enum.Enum + Pydantic 校验，禁止在业务代码里散落字符串判断。
[Default] 启用 model_config = ConfigDict(str_strip_whitespace=True) 自动去除字符串首尾空格。
[Caution] 响应模型不要复用 ORM 模型类，必须独立定义，防止数据库字段变更直接影响 API 契约。

代码示例

# app/schemas/user.py
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, ConfigDict, EmailStr, Field
 
 
class UserStatus(str, Enum):
    ACTIVE = "active"
    DISABLED = "disabled"
 
 
class UserBase(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    email: EmailStr
    status: UserStatus = UserStatus.ACTIVE
 
 
class UserCreate(UserBase):
    password: str = Field(min_length=8, max_length=128)
 
 
class UserUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=50)
    status: UserStatus | None = None
 
 
class UserOut(UserBase):
    model_config = ConfigDict(from_attributes=True)
 
    id: int
    created_at: datetime
    # 注意：不包含 password 字段，自动过滤

反模式

# 反模式：直接用 dict 接收，无校验
@router.post("/users")
def create_user(payload: dict):
    return {"id": 1, **payload}  # 无字段约束、无类型校验、可能泄露敏感字段
 
# 反模式：请求响应复用同一个模型
class User(BaseModel):
    id: int
    name: str
    password: str  # 响应时泄露密码

依赖注入

规则

[Requirement] 数据库会话、配置、鉴权用户等横切依赖必须通过 Depends() 注入，禁止在函数内部手动 new 或全局单例。
[Requirement] get_db 依赖必须使用 yield 语法，确保请求结束时会话自动关闭/回滚。
[Standard] Service/Repository 类实现为可注入依赖（构造函数参数可被 FastAPI 解析），通过 Depends(UserService) 注入，避免在路由内 service = UserService(db)。
[Standard] 鉴权依赖（如 get_current_user）在需要登录的路由上统一注入，禁止每个路由手动解析 token。
[Default] 复用的依赖用 Depends(dep, use_cache=True)（默认即开启），同请求内只执行一次。
[Suggestion] 跨多个路由的公共依赖（如分页参数、租户上下文）封装为独立依赖函数，减少重复代码。

代码示例

# app/db/session.py
from collections.abc import Generator
from sqlalchemy.orm import Session
 
def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# app/core/security.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
 
def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = decode_token(token)
    if not payload:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return get_user_by_id(payload["sub"])

# app/services/user_service.py
class UserService:
    def __init__(self, db: Session = Depends(get_db)) -> None:
        self.db = db
 
    def get_by_id(self, user_id: int) -> User | None:
        return self.db.query(User).get(user_id)

# 路由中使用
@router.get("/me", response_model=UserOut)
def get_me(current_user: User = Depends(get_current_user)) -> UserOut:
    return current_user

反模式

# 反模式：在路由内手动创建会话与服务，未注入
@router.get("/users/{user_id}")
def get_user(user_id: int):
    db = SessionLocal()               # 未注入、无关闭保证
    service = UserService(db)
    user = service.get_by_id(user_id)
    db.close()                        # 异常时不会执行
    return user
 
# 反模式：每个路由手动解析 token
@router.get("/me")
def get_me(authorization: str = Header(...)):
    token = authorization.split(" ")[1]  # 重复解析逻辑
    ...

中间件

规则

[Requirement] 全局中间件用于横切关注点（请求日志、CORS、异常兜底、请求 ID 注入），业务逻辑禁止写在中间件里。
[Requirement] CORS 必须显式配置允许的源、方法、头，生产环境禁止 allow_origins=["*"] 配合凭证。
[Standard] 使用 @app.middleware("http") 或自定义 BaseHTTPMiddleware 实现中间件，优先用官方中间件类。
[Standard] 中间件内对 request.state 注入上下文（如 request_id），通过 request.url 记录访问日志，禁止用 print。
[Default] 请求日志中间件记录：请求 ID、方法、路径、状态码、耗时；敏感请求体（密码、token）禁止记录。
[Caution] 中间件执行顺序很重要：CORS 中间件必须在最外层，异常处理中间件在日志中间件内侧，确保异常被正确转换并记录。
[Caution] 中间件内禁止做数据库写入或耗时操作，避免拖慢所有请求。

代码示例

# app/main.py
import time
import uuid
 
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
 
from app.core.config import settings
 
app = FastAPI(title=settings.app_name)
 
# 1. CORS 最外层
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)
 
 
# 2. 请求 ID + 耗时日志中间件
@app.middleware("http")
async def request_logger(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Request-ID"] = request_id
    logger.info(
        "request",
        request_id=request_id,
        method=request.method,
        path=request.url.path,
        status=response.status_code,
        elapsed_ms=round(elapsed * 1000, 2),
    )
    return response

# app/core/exceptions.py — 全局异常处理器
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
 
class BizError(Exception):
    def __init__(self, code: int, message: str) -> None:
        self.code = code
        self.message = message
 
def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(BizError)
    async def biz_error_handler(_: Request, exc: BizError) -> JSONResponse:
        return JSONResponse(
            status_code=200,
            content={"code": exc.code, "message": exc.message, "data": None},
        )

反模式

# 反模式：生产环境 CORS 全开 + 允许凭证
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],        # 安全风险
    allow_credentials=True,     # 与 * 冲突，浏览器会拒绝
    allow_methods=["*"],
    allow_headers=["*"],
)
 
# 反模式：中间件写业务逻辑 + 打印敏感信息
@app.middleware("http")
async def bad_middleware(request, call_next):
    body = await request.body()
    print(body)                 # 泄露密码/token，且消耗请求体导致后续无法读取
    if request.url.path == "/admin":
        ...                     # 业务逻辑不应在中间件
    return await call_next(request)