!文章内容如有错误或排版问题,请提交反馈,非常感谢!
在FastAPI中,请求体(RequestBody)用于接收客户端发送的结构化数据(如JSON),通常与Pydantic模型结合实现数据验证和自动文档生成。
请求体基础用法
定义Pydantic模型
from pydantic import BaseModel class ItemCreate(BaseModel): name: str price: float description: str | None = None # 可选字段 tags: list[str] = [] # 默认空列表
在路由中使用模型
from fastapi import FastAPI app = FastAPI() @app.post("/items/") async def create_item(item: ItemCreate): return {"item": item.dict()}
请求示例:
{ "name": "Laptop", "price": 5999.99, "description": "高性能笔记本", "tags": ["electronics","sale"] }
- 自动验证:缺失必填字段或类型错误会触发HTTP422错误
- Swagger文档:自动生成交互式请求示例
高级请求体配置
多请求体参数
@app.put("/items/{item_id}") async def update_item( item_id: int, # 路径参数 item: ItemCreate, # 请求体模型 q: str | None = None # 查询参数 ): return {"item_id": item_id, **item.dict(), "q": q}
单请求体多模型
from pydantic import BaseModel, EmailStr class UserBase(BaseModel): email: EmailStr class UserCreate(UserBase): password: str class UserUpdate(UserBase): name: str | None is_active: bool = True
文件上传与表单数据
文件上传
from fastapi import UploadFile @app.post("/upload/") async def upload_file( file: UploadFile, description: str = Form(...) ): return { "filename": file.filename, "size": len(await file.read()), "description": description }
混合表单与JSON
from fastapi import Form from pydantic import BaseModel class ItemForm(BaseModel): name: str price: float = Form(..., gt=0) @app.post("/form-items/") async def form_item(item: ItemForm = Depends()): return item
Pydantic模型
Pydantic模型是FastAPI生态的核心组件,用于数据验证、序列化和文档生成。通过合理运用Pydantic模型,可以实现:
- 数据验证:确保输入数据符合业务规则
- 自动文档:生成准确的API文档
- 类型安全:减少运行时类型错误
- 序列化控制:灵活处理数据输入输出格式
结合FastAPI的依赖注入系统,能够构建出健壮且易维护的API服务。
基础模型定义
必填字段与可选字段
from pydantic import BaseModel from typing import Optional class User(BaseModel): username: str # 必填字段 email: Optional[str] = None # 可选字段 age: int = 18 # 带默认值的必填字段
字段类型增强
from pydantic import HttpUrl, PaymentCardNumber class Profile(BaseModel): website: HttpUrl # 自动验证URL格式 phone: str = Field(regex=r"^1[3-9]\d{9}$") # 正则验证手机号 credit_card: PaymentCardNumber # 专用类型验证
严格类型校验
使用 strict 参数禁止类型自动转换(如字符串”123″转整数123)。
示例:严格模式
from pydantic import BaseModel, StrictInt class Data(BaseModel): number: StrictInt # 必须严格为整数类型
验证机制详解
内置验证器
通过继承 BaseModel 定义数据模型,字段类型和默认值决定了验证规则。Pydantic支持丰富的字段类型和内置校验。
常见字段类型
类型 | 说明 | 示例 |
str | 字符串 | name: str |
int | 整数 | age: int |
float | 浮点数 | price: float |
bool | 布尔值 | is_active: bool |
EmailStr | 符合邮箱格式的字符串 | email: EmailStr |
constr | 带约束的字符串 | password: constr(min_length=8) |
conint | 带约束的整数 | age: conint(ge=18) |
list, dict | 列表或字典 | tags: list[str] |
from pydantic import Field class Product(BaseModel): name: str = Field(min_length=2, max_length=50) price: float = Field(gt=0, le=1000000) stock: int = Field(ge=0)
自定义验证器
使用 @validator 装饰器定义字段级别的自定义验证逻辑。
示例:验证密码强度
from pydantic import BaseModel, validator class User(BaseModel): password: str @validator("password") def validate_password(cls, value): if len(value) < 8: raise ValueError("密码长度必须≥8") if not any(c.isupper() for c in value): raise ValueError("密码必须包含大写字母") return value
说明:
- @validator(“password”)表示验证 password 字段。
- 若验证失败,抛出ValueError,错误信息会返回给调用方。
根验证器
对整个模型的数据进行验证,可访问所有字段的值。
示例:验证字段一致性
from pydantic import BaseModel, root_validator class UserRegistration(BaseModel): password: str confirm_password: str @root_validator def check_passwords_match(cls, values): if values.get("password") != values.get("confirm_password"): raise ValueError("两次密码不一致") return values
模型继承体系
嵌套模型
模型可以嵌套其他模型,处理复杂数据结构。
示例:嵌套模型验证
class Address(BaseModel): city: str street: str class User(BaseModel): name: str address: Address # 嵌套Address模型
验证逻辑:
- address字段必须是一个包含 city 和 street 的字典。
基础模型复用
class BaseUser(BaseModel): email: EmailStr class UserCreate(BaseUser): password: str class UserResponse(BaseUser): id: int is_active: bool
联合类型
from typing import Union class Cat(BaseModel): type: str = "cat" meow: bool class Dog(BaseModel): type: str = "dog" bark: str @app.post("/animals/") async def adopt_animal(animal: Union[Cat, Dog]): return animal
混合继承
class TimestampMixin(BaseModel): created_at: datetime updated_at: datetime class Article(TimestampMixin): title: str content: str
高级模型配置
ORM模式配置
class UserORM(BaseModel): id: int name: str class Config: orm_mode = True # 允许从ORM对象转换 # 使用示例 user_orm = UserORM.from_orm(database_user)
字段别名处理
class DataModel(BaseModel): internal_name: str = Field(..., alias="externalName") class Config: allow_population_by_field_name = True # 允许通过原始名称或别名赋值
敏感数据处理
class SecureData(BaseModel): password: str class Config: validate_assignment = True # 赋值时触发验证 extra = "forbid" # 禁止额外字段 json_encoders = { SecretStr: lambda v: v.get_secret_value() if v else None }
模型配置(Config类)
自定义模型行为:
class User(BaseModel): name: str class Config: allow_mutation = False # 禁止修改实例字段 anystr_strip_whitespace = True # 自动去除字符串两端空格
自定义错误信息
通过 Field 类的参数自定义错误信息。
示例:自定义错误提示
from pydantic import BaseModel, Field class User(BaseModel): age: int = Field(..., gt=0, description="用户年龄", example=25, error_messages={ "gt": "年龄必须大于0", "type": "年龄必须是整数" })
动态模型技巧
动态默认值
使用 default_factory 生成动态默认值(如当前时间)。
示例:动态默认时间戳
from datetime import datetime from pydantic import BaseModel, Field class Post(BaseModel): title: str created_at: datetime = Field(default_factory=datetime.now)
运行时生成模型
from pydantic import create_model DynamicModel = create_model( "DynamicModel", name=(str, ...), value=(float, 0.0), __config__=dict(extra="ignore") )
条件字段模型
from typing import Literal class Payment(BaseModel): method: Literal["credit", "alipay"] card_number: str | None = None alipay_account: str | None = None @validator("card_number", always=True) def check_card(cls, v, values): if values.get("method") == "credit" and not v: raise ValueError("信用卡支付需要卡号") return v
性能优化方案
模型缓存
from functools import lru_cache @lru_cache(maxsize=100) def get_cached_model(fields: tuple): return create_model(f"CachedModel", **{f: (str, ...) for f in fields})
避免重复验证
class OptimizedModel(BaseModel): data: str def __init__(self, **data) -> None: if 'pre_validated' in data: #跳过已验证数据 super().__init__(**data) else: #执行完整验证流程 super().__init__(**data)
常见问题解决
循环引用问题
# post.py from typing import List from .user import User class Post(BaseModel): author: "User" comments: List["Comment"] # user.py from .post import Post class User(BaseModel): posts: List[Post]
时区处理
from pydantic import validator from datetime import datetime, timezone class Event(BaseModel): timestamp: datetime @validator("timestamp") def ensure_utc(cls, v): if v.tzinfo != timezone.utc: return v.astimezone(timezone.utc) return v
大数据量处理
from pydantic import parse_obj_as #直接解析原始数据避免对象实例化开销 data_list = [{"id": i, "name": f"Item {i}"} for i in range(100000)] items = parse_obj_as(list[Item], data_list)
最佳实践总结
分层设计模型
- BaseSchema: 基础字段定义
- CreateSchema: 创建专用字段
- UpdateSchema: 更新专用字段
- ResponseSchema: 响应数据结构
验证逻辑分离
class BusinessLogic: @staticmethod def validate_order(order: OrderCreate): #复杂业务规则验证 pass
文档增强
class DocumentedModel(BaseModel): """用户模型文档说明""" name: str = Field(..., description="用户姓名", example="张三") age: int = Field(description="年龄范围18-100", example=25)
性能监控
import time from pydantic import BaseModel class BenchmarkModel(BaseModel): def __init__(self, **data): start = time.perf_counter() super().__init__(**data) print(f"模型初始化耗时: {time.perf_counter()-start:.4f}s")
参考链接: