Become a sponsor

使用 Pydantic v2 进行请求数据校验。所有请求体参数通过 Pydantic BaseModel 定义,FastAPI 在调用 Endpoint 前自动完成解析和验证。
核心原则
Field(...) 定义@field_validator| 类型 | 命名格式 | 示例 |
|---|---|---|
| 创建/编辑表单 | {模块}Form | PositionForm |
| 状态更新表单 | {模块}StatusForm | PositionStatusForm |
| 查询参数表单 | {模块}QueryForm | PositionQueryForm |
继承 BaseForm
创建/编辑表单继承 core.base_schemas.BaseForm,自动获得 id: Optional[int] 字段(编辑时必填)。
以岗位模块为例,src/modules/system/position/schemas.py 完整内容:
from typing import Optional
from pydantic import BaseModel, Field
from core.base_schemas import BaseForm
class PositionForm(BaseForm):
"""岗位创建/编辑表单"""
name: str = Field(..., min_length=1, max_length=150, description="岗位名称")
status: int = Field(..., ge=1, le=2, description="岗位状态:1-在用 2-停用")
sort: int = Field(..., ge=0, le=99999, description="岗位排序")
class PositionStatusForm(BaseModel):
"""岗位状态更新表单"""
id: int = Field(..., gt=0, description="岗位ID")
status: int = Field(..., ge=1, le=2, description="岗位状态:1-在用 2-停用")BaseForm 基类
BaseForm(定义在 src/core/base_schemas.py)提供 id: Optional[int] 字段和 empty_str_to_none 验证器,自动将前端传入的空字符串 ''/'null'/'undefined' 转为 None,避免 Pydantic 整数解析报错。
| 约束 | 类型 | 说明 | 示例 |
|---|---|---|---|
min_length | str | 最小长度 | Field(..., min_length=1) |
max_length | str | 最大长度 | Field(..., max_length=150) |
ge | int/float | 大于等于 | Field(..., ge=0) |
le | int/float | 小于等于 | Field(..., le=99999) |
gt | int/float | 大于 | Field(..., gt=0) |
lt | int/float | 小于 | Field(..., lt=100) |
description | any | 字段描述(Swagger 文档) | Field(..., description="名称") |
Field 第一个参数
...:必填字段None:可选字段from pydantic import BaseModel, Field, field_validator
class UserForm(BaseForm):
"""用户创建/编辑表单"""
username: str = Field(..., min_length=3, max_length=50, description="用户名")
email: str = Field(..., max_length=100, description="邮箱")
phone: Optional[str] = Field(None, max_length=20, description="手机号")
@field_validator('username')
@classmethod
def validate_username(cls, v):
"""用户名只允许字母、数字、下划线"""
import re
if not re.match(r'^[a-zA-Z0-9_]+$', v):
raise ValueError('用户名只允许字母、数字、下划线')
return v
@field_validator('email')
@classmethod
def validate_email(cls, v):
"""邮箱格式校验"""
import re
if not re.match(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$', v):
raise ValueError('邮箱格式不正确')
return v
@field_validator('phone')
@classmethod
def validate_phone(cls, v):
"""手机号格式校验(可选字段,有值时校验)"""
if v is not None and v != '':
import re
if not re.match(r'^1[3-9]\d{9}$', v):
raise ValueError('手机号格式不正确')
return vvalidator 规则
@field_validator('field_name') 装饰器(Pydantic v2 语法)@classmethodValueError(错误消息)None 值在 Schema 的 field_validator 中查询数据库:
from pydantic import BaseModel, Field, field_validator
from core.database import get_db_session
from modules.system.position.models import Position
class PositionForm(BaseForm):
"""岗位创建/编辑表单"""
name: str = Field(..., min_length=1, max_length=150, description="岗位名称")
@field_validator('name')
@classmethod
def validate_name_unique(cls, v, info):
"""岗位名称唯一性校验"""
db = get_db_session()
query = db.query(Position).filter(
Position.name == v,
Position.is_delete == 0
)
# 编辑时排除自身
if info.data.get('id'):
query = query.filter(Position.id != info.data['id'])
if query.first():
raise ValueError('岗位名称不能重复')
return v唯一性校验位置
简单模块的唯一性校验可在 Schema 中完成。复杂模块(如用户、角色)的唯一性校验在 Service 层的 _check_unique 方法中统一处理,通过 unique_fields 字典声明。
@router.post('/add', summary='添加岗位')
@permission_required("sys:position:add")
@check_demo
async def add(request: Request, data: PositionForm):
"""
新增岗位
:param request: FastAPI请求对象
:param data: 岗位表单数据(Pydantic 自动校验)
"""
return await position_service.add(request, data)数据访问
Service 层通过 data.field 直接访问字段,或通过 data.model_dump() 获取完整字典。Pydantic 在进入 Endpoint 之前已完成校验,Service 无需重复验证。
Pydantic 校验失败时,由全局 RequestValidationError 处理器自动转换:
{
"code": 1,
"data": null,
"msg": "name: 字段不能为空 / status: 输入值超出范围",
"ok": false
}Pydantic Schema 规范要求:表单类命名 {Module}Form,继承 BaseForm;字段约束用 Field(...) 定义长度/范围;自定义校验用 @field_validator 抛 ValueError;唯一性校验在 Schema 或 Service 的 unique_fields 中完成。FastAPI 自动解析和验证,Service 直接使用 data.field 访问已校验的数据。