FastAPI 中所有常见参数校验场景
·
FastAPI 中所有常见参数校验场景 + 正确写法示例(基于 Pydantic v2 + FastAPI ≥ 0.100),涵盖:
- 路径参数(Path)
- 查询参数(Query)
- 请求体(Body / BaseModel)
- 表单(Form)
- 文件上传(File)
- 头部(Header)
- Cookie
- 嵌套模型、列表、枚举、自定义校验等
✅ 总原则(Pydantic v2 + FastAPI)
- 使用
typing.Optional[T]或T | None表示可选 - 使用
Field(default=..., ...)设置默认值和元数据 - 使用
Annotated是可选的,但能提升类型提示清晰度 - 所有验证规则(
ge,le,min_length,pattern等)都通过Field传入
1️⃣ 路径参数(Path)
场景:/items/{item_id}
from fastapi import FastAPI, Path
from typing import Annotated
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(
item_id: Annotated[int, Path(gt=0, le=1000, description="Item ID must be between 1 and 1000")]
):
return {"item_id": item_id}
✅ 支持:gt, ge, lt, le, description, example 等
2️⃣ 查询参数(Query)
场景:分页、过滤、搜索
from fastapi import Query
from typing import Annotated, Optional
@app.get("/items/")
async def list_items(
skip: Annotated[int, Query(ge=0, description="Skip N items")] = 0,
limit: Annotated[int, Query(ge=1, le=100, description="Max items per page")] = 10,
q: Annotated[Optional[str], Query(min_length=2, max_length=50, description="Search keyword")] = None,
category: Annotated[Optional[str], Query(pattern=r"^[a-z]+$")] = None, # 只允许小写字母
):
return {"skip": skip, "limit": limit, "q": q, "category": category}
⚠️ 注意:
len是错误写法,正确是le(less than or equal)
3️⃣ 请求体(Body)— 最常用!
3.1 基础可选字段(你的场景)
from pydantic import BaseModel, Field
from typing import Optional, Annotated
class SearchData(BaseModel):
type: Annotated[Optional[str], Field(None, description="笔记类型", min_length=1)] = None
is_show: Annotated[Optional[str], Field(None, description="是否显示", pattern=r"^[01]$")] = None
tags: Annotated[Optional[list[str]], Field(None, max_items=5)] = None
3.2 必填字段 + 校验
class CreateUser(BaseModel):
username: Annotated[str, Field(min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$")]
email: Annotated[str, Field(pattern=r"^\S+@\S+\.\S+$")]
age: Annotated[int, Field(ge=13, le=120)]
3.3 枚举(Literal)
from typing import Literal
class FilterRequest(BaseModel):
status: Annotated[Optional[Literal["active", "inactive", "pending"]], Field(None)] = None
sort_by: Annotated[Literal["created_at", "name"], Field("created_at")] = "created_at"
3.4 嵌套模型
class Address(BaseModel):
city: str
country: str
class UserCreate(BaseModel):
name: str
address: Optional[Address] = None # 整个嵌套对象可选
4️⃣ 表单数据(Form)
from fastapi import Form
@app.post("/login")
async def login(
username: Annotated[str, Form(min_length=3)],
password: Annotated[str, Form(min_length=8)],
remember_me: Annotated[bool, Form()] = False # checkbox 默认 False
):
return {"username": username, "remember": remember_me}
⚠️ 需要安装
bashpython-multipart:编辑
pip install python-multipart
5️⃣ 文件上传(File)
from fastapi import File, UploadFile
from typing import Annotated
@app.post("/upload")
async def upload_file(
file: Annotated[UploadFile, File(description="Image file (max 5MB)")],
description: Annotated[Optional[str], Form()] = None
):
return {
"filename": file.filename,
"content_type": file.content_type,
"description": description
}
限制文件大小需手动检查 file.size。
6️⃣ Header / Cookie
from fastapi import Header, Cookie
@app.get("/headers")
async def read_headers(
user_agent: Annotated[Optional[str], Header()] = None,
session_id: Annotated[str, Cookie()] = "default_session"
):
return {"User-Agent": user_agent, "session": session_id}
7️⃣ 自定义校验(Validator)
场景:is_show 只能是 "0" 或 "1"
from pydantic import field_validator
class SearchData(BaseModel):
is_show: Optional[str] = None
@field_validator('is_show')
@classmethod
def validate_is_show(cls, v):
if v is not None and v not in ("0", "1"):
raise ValueError("is_show must be '0' or '1'")
return v
💡 Pydantic v2 用
@field_validator替代了 v1 的@validator
8️⃣ 列表 & 字典校验
class BatchUpdate(BaseModel):
ids: Annotated[list[int], Field(min_items=1, max_items=100)]
metadata: dict[str, str] # 键值都必须是字符串
9️⃣ 时间字段
from datetime import datetime
class EventCreate(BaseModel):
start_time: datetime
end_time: Annotated[datetime, Field()]
@field_validator('end_time')
@classmethod
def end_after_start(cls, v, info):
start = info.data.get('start_time')
if start and v <= start:
raise ValueError("end_time must be after start_time")
return v
🔟 路由中组合使用
@app.post("/search")
async def search(
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(le=50)] = 10,
filters: SearchData, # 来自 Body
x_trace_id: Annotated[Optional[str], Header()] = None,
db: Session = Depends(get_db)
):
# 业务逻辑
pass
📌 关键总结表
表格
| 参数类型 | 如何声明可选 | 默认值 | 常用校验 |
|---|---|---|---|
| Path | 不支持可选(路径必须存在) | — | gt, le, regex |
| Query | Optional[T] + = default |
= None 或 = 0 |
ge, le, min_length, pattern |
| Body | Optional[T] + Field(None) |
= None |
同上 + validator |
| Form | 同 Body | = None |
同上 |
| File | UploadFile | None |
= None |
手动检查 size/type |
| Header/Cookie | Optional[str] |
= None |
一般不校验 |
更多推荐
所有评论(0)