5.3 KiB
5.3 KiB
id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week, order
| id | title | type | domain | stack | tags | level | prerequisites | est_hours | related | week | order | |||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| python.fastapi.routing-validation | 路由和参数校验：路径 / 查询 / 请求体、pydantic 模型、响应模型 | lesson | python |
|
|
basic |
|
2 |
|
|
1 |
路由和参数校验：路径 / 查询 / 请求体、pydantic 模型、响应模型
解释说明
- 参数来源由类型和位置决定：路径里出现的参数 → 路径参数；简单类型（int、str）→ 查询参数；pydantic 模型 → JSON 请求体；
Header()、Cookie()显式声明。 - 校验：
Field(ge=1, le=100, min_length=1, pattern=...)、Query(...)、Path(...)；Literal限定枚举值；校验失败自动返回 422，带出错字段和原因。 - 响应模型：返回类型标注或
response_model=，会过滤掉模型之外的字段——防止把密码哈希等内部字段返回出去。 - 自动文档：启动后访问
/docs（Swagger UI），前端联调直接用。 async def还是def：路由里调用的都是异步库 →async def；调用同步阻塞库 → 普通def（FastAPI 会放到线程池执行）。不要在async def里调用阻塞函数。- APIRouter：按模块拆分路由，
app.include_router(router, prefix="/v1/docs")。
和 Gin 对照：c.Param ↔ 路径参数，c.Query ↔ 查询参数，ShouldBindJSON + binding 标签 ↔ pydantic 模型 + Field。
案例
知识库的文档接口：创建文档（标题 1200 字、标签最多 10 个、可见范围只能是 100）。用户传了 private / team / public）、按 ID 查询、分页列表（页码 ≥ 1，每页 1"visibility": "everyone"，FastAPI 自动返回 422 和具体字段，不用手写校验代码。
代码示例
from datetime import datetime, timezone
from typing import Annotated, Literal
from fastapi import APIRouter, FastAPI, HTTPException, Path, Query
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
class DocCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
content: str = Field(min_length=1)
tags: list[str] = Field(default_factory=list, max_length=10)
visibility: Literal["private", "team", "public"] = "team"
class DocOut(BaseModel): # 响应模型：只暴露这些字段
id: int
title: str
tags: list[str]
visibility: str
created_at: datetime
class Page(BaseModel):
items: list[DocOut]
total: int
router = APIRouter(prefix="/v1/docs", tags=["文档"])
DB: dict[int, dict] = {}
@router.post("", status_code=201)
async def create_doc(body: DocCreate) -> DocOut:
doc_id = len(DB) + 1
DB[doc_id] = {**body.model_dump(), "id": doc_id, "owner_secret": "内部字段",
"created_at": datetime.now(timezone.utc)}
return DocOut(**DB[doc_id]) # 只取模型里声明的字段
@router.get("/{doc_id}")
async def get_doc(doc_id: Annotated[int, Path(ge=1)]) -> DocOut:
if doc_id not in DB:
raise HTTPException(status_code=404, detail="文档不存在")
return DocOut(**DB[doc_id])
@router.get("")
async def list_docs(page: Annotated[int, Query(ge=1)] = 1,
size: Annotated[int, Query(ge=1, le=100)] = 20,
tag: str | None = None) -> Page:
docs = [d for d in DB.values() if tag is None or tag in d["tags"]]
start = (page - 1) * size
return Page(items=[DocOut(**d) for d in docs[start:start + size]], total=len(docs))
app = FastAPI()
app.include_router(router)
c = TestClient(app)
r = c.post("/v1/docs", json={"title": "年假制度", "content": "……", "tags": ["hr"]})
print(r.status_code, r.json()["id"], "owner_secret" in r.json()) # 201 1 False
r = c.post("/v1/docs", json={"title": "", "content": "x", "visibility": "everyone"})
print(r.status_code, [e["loc"] for e in r.json()["detail"]]) # 422 [['body', 'title'], ['body', 'visibility']]
print(c.get("/v1/docs", params={"size": 500}).status_code) # 422
print(c.get("/v1/docs/99").json()) # {'detail': '文档不存在'}
print(c.get("/v1/docs", params={"tag": "hr"}).json()["total"]) # 1
易错点
- 在
async def路由里调用同步数据库驱动或requests：阻塞事件循环，吞吐骤降。 - 直接返回 ORM 对象或字典而不声明响应模型：内部字段（密码哈希、内部备注）泄露。
- 可变默认值：
tags: list[str] = []在 pydantic 里是安全的（会复制），但养成default_factory的习惯更清晰。 - 路由顺序：
/docs/{doc_id}写在/docs/stats之前，stats会被当成 doc_id 去解析，返回 422。固定路径放在前面。 - 422 的错误格式直接给前端：默认格式偏技术化，通常用异常处理器转换成团队统一的格式（见下一课）。
- pydantic v1 和 v2 写法混用：v2 用
model_dump()、model_validate()、field_validator；网上很多旧教程是 v1 写法（dict()、validator）。
面试常问
见 FastAPI 面试题：FastAPI 为什么快、参数校验原理、async def 和 def 的区别、响应模型的作用。