190 lines
6.3 KiB
Markdown
190 lines
6.3 KiB
Markdown
# Pydantic Schema规范
|
|
|
|
## 概述
|
|
|
|
使用 Pydantic v2 进行请求数据校验。所有请求体参数通过 Pydantic `BaseModel` 定义，FastAPI 在调用 Endpoint 前自动完成解析和验证。
|
|
|
|
:::tip 核心原则
|
|
- 请求体参数必须声明为 Pydantic 模型（Form 类）
|
|
- 字段约束使用 `Field(...)` 定义
|
|
- 自定义校验使用 `@field_validator`
|
|
- 唯一性校验在 Schema 中查询数据库
|
|
:::
|
|
|
|
## 命名规范
|
|
|
|
| 类型 | 命名格式 | 示例 |
|
|
|------|---------|------|
|
|
| 创建/编辑表单 | `{模块}Form` | `PositionForm` |
|
|
| 状态更新表单 | `{模块}StatusForm` | `PositionStatusForm` |
|
|
| 查询参数表单 | `{模块}QueryForm` | `PositionQueryForm` |
|
|
|
|
:::tip 继承 BaseForm
|
|
创建/编辑表单继承 `core.base_schemas.BaseForm`，自动获得 `id: Optional[int]` 字段（编辑时必填）。
|
|
:::
|
|
|
|
## PositionForm 示例
|
|
|
|
以岗位模块为例，`src/modules/system/position/schemas.py` 完整内容：
|
|
|
|
```python
|
|
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-停用")
|
|
```
|
|
|
|
:::tip BaseForm 基类
|
|
`BaseForm`（定义在 `src/core/base_schemas.py`）提供 `id: Optional[int]` 字段和 `empty_str_to_none` 验证器，自动将前端传入的空字符串 `''`/`'null'`/`'undefined'` 转为 `None`，避免 Pydantic 整数解析报错。
|
|
:::
|
|
|
|
## Field 约束
|
|
|
|
| 约束 | 类型 | 说明 | 示例 |
|
|
|------|------|------|------|
|
|
| `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="名称")` |
|
|
|
|
:::tip Field 第一个参数
|
|
- `...`：必填字段
|
|
- `None`：可选字段
|
|
- 具体值：默认值
|
|
:::
|
|
|
|
## field_validator 自定义校验
|
|
|
|
```python
|
|
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 v
|
|
```
|
|
|
|
:::tip validator 规则
|
|
- 使用 `@field_validator('field_name')` 装饰器（Pydantic v2 语法）
|
|
- 必须加 `@classmethod`
|
|
- 校验失败抛出 `ValueError(错误消息)`
|
|
- 可选字段需判断 `None` 值
|
|
:::
|
|
|
|
## 唯一性校验
|
|
|
|
在 Schema 的 `field_validator` 中查询数据库：
|
|
|
|
```python
|
|
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
|
|
```
|
|
|
|
:::tip 唯一性校验位置
|
|
简单模块的唯一性校验可在 Schema 中完成。复杂模块（如用户、角色）的唯一性校验在 Service 层的 `_check_unique` 方法中统一处理，通过 `unique_fields` 字典声明。
|
|
:::
|
|
|
|
## 在 Endpoint 中使用
|
|
|
|
```python
|
|
@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)
|
|
```
|
|
|
|
:::tip 数据访问
|
|
Service 层通过 `data.field` 直接访问字段，或通过 `data.model_dump()` 获取完整字典。Pydantic 在进入 Endpoint 之前已完成校验，Service 无需重复验证。
|
|
:::
|
|
|
|
## 错误响应
|
|
|
|
Pydantic 校验失败时，由全局 `RequestValidationError` 处理器自动转换：
|
|
|
|
```json
|
|
{
|
|
"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` 访问已校验的数据。
|