FastAPI模式Skill py-fastapi-patterns

FastAPI 模式

问题陈述

FastAPI API设计直接影响前端。这里的不良模式会导致前端错误、糟糕的开发者体验和集成问题。OpenAPI模式驱动前端代码生成。

---

模式：依赖注入

**问题：**对于认证、会话和服务的重复代码。

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

# ✅ 正确：常用需求的依赖
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_session),
) -> User:
    user = await verify_token_and_get_user(token, session)
    if not user:
        raise HTTPException(401, "无效认证")
    return user

async def get_current_active_user(
    user: User = Depends(get_current_user),
) -> User:
    if not user.is_active:
        raise HTTPException(403, "用户不活跃")
    return user

# ✅ 正确：使用依赖的端点
@router.post("/assessments", response_model=AssessmentRead)
async def create_assessment(
    data: AssessmentCreate,
    current_user: User = Depends(get_current_active_user),
    session: AsyncSession = Depends(get_session),
) -> AssessmentRead:
    assessment = Assessment(**data.model_dump(), user_id=current_user.id)
    session.add(assessment)
    await session.commit()
    await session.refresh(assessment)
    return assessment

依赖链： get_session → get_current_user → get_current_active_user

---

模式：响应模型

**问题：**不一致的响应、暴露内部字段、糟糕的OpenAPI文档。

# ✅ 正确：明确的response_model
@router.get("/users/{user_id}", response_model=UserRead)
async def get_user(
    user_id: UUID,
    session: AsyncSession = Depends(get_session),
) -> UserRead:
    user = await get_user_or_404(user_id, session)
    return user  # 自动过滤到UserRead字段

# ✅ 正确：列表响应
@router.get("/users", response_model=list[UserRead])
async def list_users(...) -> list[UserRead]:
    ...

# ✅ 正确：分页响应
class PaginatedResponse(SQLModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    size: int

@router.get("/assessments", response_model=PaginatedResponse[AssessmentRead])
async def list_assessments(...):
    ...

# ❌ 错误：没有response_model（暴露所有内容）
@router.get("/users/{user_id}")
async def get_user(user_id: UUID) -> User:  # 暴露hashed_password!
    ...

为什么response_model很重要：

1. 只过滤指定字段的输出
2. 生成准确的OpenAPI模式
3. 前端Orval代码生成依赖于此

---

模式：错误处理

**问题：**不一致的错误响应，缺少上下文。

from fastapi import HTTPException, status

# ✅ 正确：特定的HTTP异常
@router.get("/assessments/{id}")
async def get_assessment(id: UUID, session: AsyncSession = Depends(get_session)):
    result = await session.execute(
        select(Assessment).where(Assessment.id == id)
    )
    assessment = result.scalar_one_or_none()
    
    if not assessment:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Assessment {id} not found",
        )
    
    return assessment

# ✅ 正确：带有处理器的自定义异常
class AssessmentNotFoundError(Exception):
    def __init__(self, assessment_id: UUID):
        self.assessment_id = assessment_id

@app.exception_handler(AssessmentNotFoundError)
async def assessment_not_found_handler(request: Request, exc: AssessmentNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "detail": f"Assessment {exc.assessment_id} not found",
            "error_code": "ASSESSMENT_NOT_FOUND",
        },
    )

# ✅ 正确：验证错误详情
@router.post("/assessments")
async def create_assessment(data: AssessmentCreate):
    if data.end_date < data.start_date:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="end_date must be after start_date",
        )

HTTP状态代码：

代码	用途
200	成功的GET、PUT、PATCH
201	成功的POST（已创建）
204	成功的DELETE（无内容）
400	错误的请求（格式错误）
401	未授权（未认证）
403	禁止（已认证但不允许）
404	未找到
422	验证错误
500	服务器错误

---

模式：路由排序

**问题：**FastAPI匹配第一个路由。排序对于重叠路径很重要。

# ❌ 错误：通用路由在特定路由之前
@router.get("/users/{user_id}")  # 这会捕获"me"作为user_id!
async def get_user(user_id: str):
    ...

@router.get("/users/me")  # 从未到达
async def get_current_user():
    ...

# ✅ 正确：特定路由在通用之前
@router.get("/users/me")  # 特定第一
async def get_current_user():
    ...

@router.get("/users/{user_id}")  # 通用在后
async def get_user(user_id: UUID):  # UUID类型也有帮助
    ...

**记住：**总是先定义特定路由，然后是通用参数化路由。

---

模式：路径和查询参数

# 路径参数 - 必需的，URL的一部分
@router.get("/users/{user_id}")
async def get_user(user_id: UUID):  # /users/123
    ...

# 查询参数 - 可选的，在?之后
@router.get("/assessments")
async def list_assessments(
    status: str | None = None,        # /assessments?status=active
    skip: int = 0,                     # /assessments?skip=10
    limit: int = Query(default=20, le=100),  # 带验证
):
    ...

# 枚举用于受限值
class AssessmentStatus(str, Enum):
    DRAFT = "draft"
    ACTIVE = "active"
    COMPLETED = "completed"

@router.get("/assessments")
async def list_assessments(status: AssessmentStatus | None = None):
    ...

---

模式：请求体验证

from pydantic import Field, field_validator

class AssessmentCreate(SQLModel):
    title: str = Field(min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=1000)
    skill_areas: list[str] = Field(min_length=1)
    
    @field_validator("skill_areas")
    @classmethod
    def validate_skill_areas(cls, v: list[str]) -> list[str]:
        valid_areas = {"fundamentals", "advanced", "strategy"}
        for area in v:
            if area not in valid_areas:
                raise ValueError(f"Invalid skill area: {area}")
        return v

# 自动验证 - 失败时返回422
@router.post("/assessments", response_model=AssessmentRead)
async def create_assessment(data: AssessmentCreate):
    ...

---

模式：中间件

**问题：**跨领域关注点，如日志、CORS、计时。

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

app = FastAPI()

# CORS - 顺序很重要，尽早添加
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 或["*"]用于开发
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 自定义计时中间件
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    response.headers["X-Process-Time"] = str(duration)
    return response

# 中间件顺序：最后添加=首先执行

---

模式：OpenAPI模式

**问题：**模式影响前端代码生成。保持其清晰。

from fastapi import FastAPI

app = FastAPI(
    title="我的API",
    version="1.0.0",
    description="API描述在这里",
)

# 良好的模式描述
class AssessmentCreate(SQLModel):
    """创建一个新的技能评估。"""
    
    title: str = Field(description="向用户显示的评估标题")
    skill_areas: list[str] = Field(
        description="要评估的技能领域列表",
        examples=[["fundamentals", "strategy"]],
    )

# 端点文档
@router.post(
    "/assessments",
    response_model=AssessmentRead,
    summary="创建评估",
    description="为当前用户创建一个新的技能评估。",
    responses={
        201: {"description": "评估成功创建"},
        422: {"description": "验证错误"},
    },
)
async def create_assessment(data: AssessmentCreate):
    ...

---

模式：路由器组织

# app/routers/assessments.py
from fastapi import APIRouter

router = APIRouter(
    prefix="/assessments",
    tags=["Assessments"],
)

@router.get("/")
async def list_assessments():
    ...

@router.post("/")
async def create_assessment():
    ...

# app/main.py
from app.routers import assessments, users, training

app.include_router(assessments.router, prefix="/api")
app.include_router(users.router, prefix="/api")
app.include_router(training.router, prefix="/api")

---

参考资料

• OpenAPI在/docs或/openapi.json
• FastAPI文档：https://fastapi.tiangolo.com/

---

常见问题

问题	可能的原因	解决方案
错误匹配端点	路由排序	将特定路由放在通用之前
内部字段暴露	缺少response_model	添加response_model=
有效输入422错误	Pydantic v2严格性	检查字段验证器
CORS错误	缺少/错误的中间件	首先添加CORSMiddleware
前端类型错误	模式不匹配	检查OpenAPI，重新生成API客户端

---

检测命令

# 查找缺少response_model的端点
grep -rn "@router\." --include="*.py" | grep -v "response_model"

# 查找潜在的路由排序问题
grep -rn "@router.get" --include="*.py" | grep -E '"/\w+/\{|"/\w+/\w+"'

# 检查OpenAPI模式
curl http://localhost:8000/openapi.json | jq '.paths'