117 lines
5.3 KiB
Markdown
117 lines
5.3 KiB
Markdown
---
|
||
id: python.fastapi.routing-validation
|
||
title: 路由和参数校验：路径 / 查询 / 请求体、pydantic 模型、响应模型
|
||
type: lesson
|
||
domain: python
|
||
stack: [fastapi]
|
||
tags: [路由, pydantic, 参数校验, 响应模型, OpenAPI]
|
||
level: basic
|
||
prerequisites: [python.python.typing-errors-testing]
|
||
est_hours: 2
|
||
related: [python.fastapi.deps-auth-errors, backend.gin.routing-binding, python.fastapi.interview]
|
||
week: [4]
|
||
order: 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`。
|
||
|
||
## 案例
|
||
|
||
知识库的文档接口：创建文档（标题 1~200 字、标签最多 10 个、可见范围只能是 `private / team / public`）、按 ID 查询、分页列表（页码 ≥ 1，每页 1~100）。用户传了 `"visibility": "everyone"`，FastAPI 自动返回 422 和具体字段，不用手写校验代码。
|
||
|
||
## 代码示例
|
||
|
||
```python
|
||
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 面试题](interview.md)：FastAPI 为什么快、参数校验原理、async def 和 def 的区别、响应模型的作用。
|