refactor(schema): 移除重复的模型验证逻辑并简化字段验证

统一移除各模块schema中重复的model_validator前置处理逻辑,改用更简洁的字段验证方式
优化字段验证逻辑,移除冗余的字符串处理和类型转换代码
This commit is contained in:
zhangtao
2025-11-11 23:34:30 +08:00
parent 5d0d0cd26a
commit b678588d13
15 changed files with 34 additions and 401 deletions
@@ -4,7 +4,7 @@ from typing import Optional, Dict, Any
from pydantic import ConfigDict, Field, HttpUrl, BaseModel
from app.core.base_schema import BaseSchema
from app.common.enums import McpLLMProvider, McpType
from app.common.enums import McpType
class ChatQuerySchema(BaseModel):
@@ -2,7 +2,7 @@
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from typing import Optional
import re
from app.core.base_schema import BaseSchema
from app.core.validator import DateTimeStr, datetime_validator
@@ -26,30 +26,6 @@ class JobCreateSchema(BaseModel):
description: Optional[str] = Field(default=None, max_length=255, description='描述')
status: Optional[bool] = Field(default=False, description='任务状态:启动,停止')
@model_validator(mode='before')
@classmethod
def _normalize(cls, data):
"""前置归一化:字符串去空格、布尔/数字兼容转换。"""
if isinstance(data, dict):
for key in ('name', 'func', 'trigger', 'args', 'kwargs', 'jobstore', 'executor', 'trigger_args', 'start_date', 'end_date', 'description'):
val = data.get(key)
if isinstance(val, str):
data[key] = val.strip()
for bkey in ('coalesce', 'status'):
val = data.get(bkey)
if isinstance(val, str):
lowered = val.strip().lower()
if lowered in {'true', '1', 'y', 'yes'}:
data[bkey] = True
elif lowered in {'false', '0', 'n', 'no'}:
data[bkey] = False
elif isinstance(val, int):
data[bkey] = bool(val)
val = data.get('max_instances')
if isinstance(val, str) and val.strip().isdigit():
data['max_instances'] = int(val.strip())
return data
@field_validator('trigger')
@classmethod
def _validate_trigger(cls, v: str) -> str:
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.core.base_schema import BaseSchema
from urllib.parse import urlparse
@@ -15,31 +15,6 @@ class ApplicationCreateSchema(BaseModel):
status: bool = Field(True, description="是否启用(True:启用 False:禁用)")
description: Optional[str] = Field(default=None, max_length=255, description="描述")
@model_validator(mode="before")
@classmethod
def _normalize(cls, data):
"""模型级前置处理:去除首尾空格,空字符串转为 None(可选字段),并规范布尔。"""
if isinstance(data, dict):
for key in ("name", "access_url", "icon_url", "description"):
val = data.get(key)
if isinstance(val, str):
val = val.strip()
# 将可选字段的空字符串转换为 None
if key in ("icon_url", "description") and val == "":
val = None
data[key] = val
# 规范布尔字符串/数字为布尔值
status_val = data.get("status")
if isinstance(status_val, str):
lowered = status_val.strip().lower()
if lowered in {"true", "1", "y", "yes"}:
data["status"] = True
elif lowered in {"false", "0", "n", "no"}:
data["status"] = False
elif isinstance(status_val, int):
data["status"] = bool(status_val)
return data
@field_validator('name')
@classmethod
def _validate_name_length(cls, v: str) -> str:
@@ -48,19 +48,6 @@ class ImportModel(BaseModel):
filed_info: Optional[list[ImportFieldModel]] = Field(description='字段关联表')
file_name: Optional[str] = Field(description='文件名')
@model_validator(mode='before')
@classmethod
def _normalize(cls, data):
if isinstance(data, dict):
for key in ('table_name', 'sheet_name', 'file_name'):
val = data.get(key)
if isinstance(val, str):
val = val.strip()
if val == '':
val = None
data[key] = val
return data
@model_validator(mode='after')
def _validate(self):
# excel_column 不重复(忽略 None
@@ -8,68 +8,34 @@ from app.core.base_schema import BaseSchema
class DemoCreateSchema(BaseModel):
"""新增模型"""
name: str = Field(..., max_length=50, description='名称')
name: str = Field(..., min_length=2, max_length=50, description='名称')
status: bool = Field(True, description="是否启用(True:启用 False:禁用)")
description: Optional[str] = Field(default=None, max_length=255, description="描述")
@field_validator('name')
@classmethod
def _validate_name(cls, v: str) -> str:
def validate_name(cls, v: str) -> str:
"""验证名称字段的格式和内容"""
# 去除首尾空格
v = v.strip()
if not v:
raise ValueError('名称不能为空')
return v
@model_validator(mode='before')
@classmethod
def _normalize(cls, data):
if isinstance(data, dict):
for key in ('name', 'description'):
val = data.get(key)
if isinstance(val, str):
val = val.strip()
if key == 'description' and val == '':
val = None
data[key] = val
# status兼容
val = data.get('status')
if isinstance(val, str):
lowered = val.strip().lower()
if lowered in {'true', '1', 'y', 'yes'}:
data['status'] = True
elif lowered in {'false', '0', 'n', 'no'}:
data['status'] = False
elif isinstance(val, int):
data['status'] = bool(val)
return data
@model_validator(mode='wrap')
@classmethod
def _wrap(cls, data, handler):
# 进一步处理:压缩名称/描述中的多余空白,并支持更多 status 同义词
if isinstance(data, dict):
name = data.get('name')
if isinstance(name, str):
data['name'] = ' '.join(name.split())
status_val = data.get('status')
if isinstance(status_val, str):
lowered = status_val.strip().lower()
if lowered in {'enabled', 'enable', 'on'}:
data['status'] = True
elif lowered in {'disabled', 'disable', 'off'}:
data['status'] = False
desc = data.get('description')
if isinstance(desc, str):
data['description'] = ' '.join(desc.split())
result = handler(data)
return result
@model_validator(mode='after')
def _check_disabled_requires_description(self):
# 业务示例:禁用时必须填写描述
if self.status is False and (self.description is None or (isinstance(self.description, str) and self.description.strip() == '')):
raise ValueError('禁用时必须填写描述')
def _after_validation(self):
"""
核心业务规则校验
"""
# 长度校验:名称最小长度
if len(self.name) < 2 or len(self.name) > 50:
raise ValueError('名称长度必须在2-50个字符之间')
# 格式校验:名称只能包含字母、数字、下划线和中划线
if not self.name.isalnum() and not all(c in '-_' for c in self.name):
raise ValueError('名称只能包含字母、数字、下划线和中划线')
return self
class DemoUpdateSchema(DemoCreateSchema):
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from typing import Optional, List
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.core.base_schema import BaseSchema
@@ -36,32 +36,6 @@ class DeptCreateSchema(BaseModel):
raise ValueError("部门编码必须以字母开头,且仅包含字母/数字/下划线")
return v
@model_validator(mode='before')
@classmethod
def _normalize(cls, data):
if isinstance(data, dict):
for key in ('code', 'description'):
val = data.get(key)
if isinstance(val, str):
val = val.strip()
if key == 'description' and val == '':
val = None
data[key] = val
pid = data.get('parent_id')
if isinstance(pid, str) and pid.strip().isdigit():
data['parent_id'] = int(pid.strip())
# status兼容
status_val = data.get('status')
if isinstance(status_val, str):
lowered = status_val.strip().lower()
if lowered in {'true', '1', 'y', 'yes'}:
data['status'] = True
elif lowered in {'false', '0', 'n', 'no'}:
data['status'] = False
elif isinstance(status_val, int):
data['status'] = bool(status_val)
return data
class DeptUpdateSchema(DeptCreateSchema):
"""部门更新模型"""
@@ -1,5 +1,5 @@
import re
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from typing import Optional
from app.core.base_schema import BaseSchema
@@ -54,25 +54,16 @@ class DictDataCreateSchema(BaseModel):
is_default: Optional[bool] = Field(default=None, description='是否默认(Y是 N否)')
status: Optional[bool] = Field(default=None, description='状态(1正常 0停用)')
description: Optional[str] = Field(default=None, max_length=255, description="描述")
@field_validator('dict_label')
@classmethod
def validate_dict_label(cls, value: str):
if not value or value.strip() == '':
@model_validator(mode='after')
def validate_after(self):
if self.dict_label is None or self.dict_label.strip() == '':
raise ValueError('字典标签不能为空')
return value
@field_validator('dict_value')
def validate_dict_value(cls, value: str):
if not value or value.strip() == '':
if self.dict_value is None or self.dict_value.strip() == '':
raise ValueError('字典键值不能为空')
return value
@field_validator('dict_type')
def validate_dict_type(cls, value: str):
if not value or value.strip() == '':
if self.dict_type is None or self.dict_type.strip() == '':
raise ValueError('字典类型不能为空')
return value
return self
class DictDataUpdateSchema(DictDataCreateSchema):
@@ -23,25 +23,6 @@ class OperationLogCreateSchema(BaseModel):
description: Optional[str] = Field(default=None, max_length=255, description="描述")
creator_id: Optional[int] = Field(default=None, description="创建人ID")
@model_validator(mode='before')
@classmethod
def _normalize(cls, values):
if isinstance(values, dict):
# 字符串去空格
for k in ["request_path", "request_method", "request_payload", "request_ip", "login_location", "request_os", "request_browser", "response_json", "process_time", "description"]:
if k in values and isinstance(values[k], str):
values[k] = values[k].strip() or None if values[k].strip() == "" and k in {"request_payload", "response_json", "description"} else values[k].strip()
# 方法大写
if "request_method" in values and isinstance(values["request_method"], str):
values["request_method"] = values["request_method"].strip().upper()
# 响应码转整数
if "response_code" in values and isinstance(values["response_code"], str):
try:
values["response_code"] = int(values["response_code"].strip())
except Exception:
pass
return values
@field_validator("type")
@classmethod
def _validate_type(cls, value: Optional[int]):
@@ -14,25 +14,6 @@ class NoticeCreateSchema(BaseModel):
status: bool = Field(default=True, description="是否启用(True:启用 False:禁用)")
description: Optional[str] = Field(default=None, max_length=255, description="描述")
@model_validator(mode='before')
@classmethod
def _normalize(cls, values):
if isinstance(values, dict):
# 字符串去空格
for k in ["notice_title", "notice_type", "notice_content", "description"]:
if k in values and isinstance(values[k], str):
values[k] = values[k].strip() or None if values[k].strip() == "" and k == "description" else values[k].strip()
# 布尔兼容
if "status" in values and isinstance(values["status"], str):
values["status"] = values["status"].strip().lower() in {"true", "1", "yes", "y"}
# 类型映射
mapping = {"1": "1", "2": "2", "通知": "1", "公告": "2", "notice": "1", "announcement": "2"}
if "notice_type" in values and isinstance(values["notice_type"], str):
v = values["notice_type"].strip().lower()
if v in mapping:
values["notice_type"] = mapping[v]
return values
@field_validator("notice_type")
@classmethod
def _validate_notice_type(cls, value: str):
@@ -46,8 +27,6 @@ class NoticeCreateSchema(BaseModel):
raise ValueError("公告标题不能为空")
if not self.notice_content.strip():
raise ValueError("公告内容不能为空")
if self.status is False and (not self.description or not str(self.description).strip()):
raise ValueError("禁用状态下必须填写描述")
return self
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.core.base_schema import BaseSchema
@@ -15,33 +15,6 @@ class ParamsCreateSchema(BaseModel):
status: bool = Field(default=True, description="状态(True:正常 False:停用)")
description: Optional[str] = Field(default=None, max_length=500, description="描述")
@model_validator(mode='before')
@classmethod
def _normalize(cls, data):
"""前置归一化:字符串去空格、空串转 None、布尔兼容转换,并规范键为小写。"""
if isinstance(data, dict):
for key in ('config_name', 'config_key', 'config_value', 'description'):
val = data.get(key)
if isinstance(val, str):
val = val.strip()
if key in ('config_value', 'description') and val == '':
val = None
data[key] = val
# 规范键为小写
if isinstance(data.get('config_key'), str):
data['config_key'] = data['config_key'].lower()
# 规范布尔
for bkey in ('config_type', 'status'):
val = data.get(bkey)
if isinstance(val, str):
lowered = val.strip().lower()
if lowered in {'true', '1', 'y', 'yes'}:
data[bkey] = True
elif lowered in {'false', '0', 'n', 'no'}:
data[bkey] = False
elif isinstance(val, int):
data[bkey] = bool(val)
return data
@field_validator('config_key')
@classmethod
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.core.base_schema import BaseSchema
from app.core.validator import DateTimeStr
@@ -21,33 +21,6 @@ class PositionCreateSchema(BaseModel):
raise ValueError('岗位名称不能为空')
return v
@model_validator(mode='before')
@classmethod
def _normalize(cls, data):
if isinstance(data, dict):
for key in ('name', 'description'):
val = data.get(key)
if isinstance(val, str):
val = val.strip()
if key == 'description' and val == '':
val = None
data[key] = val
# order字符串转为整数
order_val = data.get('order')
if isinstance(order_val, str) and order_val.strip().isdigit():
data['order'] = int(order_val.strip())
# status兼容
status_val = data.get('status')
if isinstance(status_val, str):
lowered = status_val.strip().lower()
if lowered in {'true', '1', 'y', 'yes'}:
data['status'] = True
elif lowered in {'false', '0', 'n', 'no'}:
data['status'] = False
elif isinstance(status_val, int):
data['status'] = bool(status_val)
return data
class PositionUpdateSchema(PositionCreateSchema):
"""岗位更新模型"""
@@ -30,38 +30,6 @@ class RoleCreateSchema(BaseModel):
raise ValueError("角色编码需字母开头,允许字母/数字/下划线,长度2-40")
return v
@field_validator("name")
@classmethod
def validate_name(cls, value: str):
v = value.strip()
if not v:
raise ValueError("角色名称不能为空")
return v
@model_validator(mode='before')
@classmethod
def _normalize(cls, values):
if isinstance(values, dict):
for k in ["name", "code", "description"]:
if k in values and isinstance(values[k], str):
values[k] = values[k].strip() or None if values[k].strip() == "" else values[k].strip()
# bool 兼容
if "status" in values and isinstance(values["status"], str):
values["status"] = values["status"].strip().lower() in {"true", "1", "yes", "y"}
# 数字兼容
if "order" in values and isinstance(values["order"], str):
try:
values["order"] = int(values["order"].strip())
except Exception:
pass
return values
@model_validator(mode='after')
def _validate_after(self):
if self.status is False and (not self.description or not str(self.description).strip()):
raise ValueError("禁用状态下必须填写描述")
return self
class RolePermissionSettingSchema(BaseModel):
"""角色权限配置模型"""
@@ -70,18 +38,6 @@ class RolePermissionSettingSchema(BaseModel):
menu_ids: List[int] = Field(default_factory=list, description='菜单ID列表')
dept_ids: List[int] = Field(default_factory=list, description='部门ID列表')
@model_validator(mode='before')
@classmethod
def _normalize(cls, values):
if isinstance(values, dict):
for k in ["role_ids", "menu_ids", "dept_ids"]:
if k in values and values[k] is not None:
try:
values[k] = list({int(x) for x in values[k]})
except Exception:
pass
return values
@model_validator(mode='after')
def validate_fields(self):
"""验证权限配置字段"""
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.core.base_schema import BaseSchema
@@ -20,57 +20,6 @@ class TenantCreateSchema(BaseModel):
raise ValueError('名称不能为空')
return v
@model_validator(mode='before')
@classmethod
def _normalize(cls, data):
if isinstance(data, dict):
for key in ('name', 'description'):
val = data.get(key)
if isinstance(val, str):
val = val.strip()
if key == 'description' and val == '':
val = None
data[key] = val
# status兼容
val = data.get('status')
if isinstance(val, str):
lowered = val.strip().lower()
if lowered in {'true', '1', 'y', 'yes'}:
data['status'] = True
elif lowered in {'false', '0', 'n', 'no'}:
data['status'] = False
elif isinstance(val, int):
data['status'] = bool(val)
return data
@model_validator(mode='wrap')
@classmethod
def _wrap(cls, data, handler):
# 进一步处理:压缩名称/描述中的多余空白,并支持更多 status 同义词
if isinstance(data, dict):
name = data.get('name')
if isinstance(name, str):
data['name'] = ' '.join(name.split())
status_val = data.get('status')
if isinstance(status_val, str):
lowered = status_val.strip().lower()
if lowered in {'enabled', 'enable', 'on'}:
data['status'] = True
elif lowered in {'disabled', 'disable', 'off'}:
data['status'] = False
desc = data.get('description')
if isinstance(desc, str):
data['description'] = ' '.join(desc.split())
result = handler(data)
return result
@model_validator(mode='after')
def _check_disabled_requires_description(self):
# 业务示例:禁用时必须填写描述
if self.status is False and (self.description is None or (isinstance(self.description, str) and self.description.strip() == '')):
raise ValueError('禁用时必须填写描述')
return self
class TenantUpdateSchema(TenantCreateSchema):
"""更新模型"""
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from typing import Optional, List
from pydantic import BaseModel, ConfigDict, Field, EmailStr, field_validator, model_validator
from pydantic import BaseModel, ConfigDict, Field, EmailStr, field_validator
from app.core.validator import DateTimeStr, mobile_validator
from app.core.base_schema import BaseSchema, CommonSchema
@@ -59,24 +59,6 @@ class UserRegisterSchema(BaseModel):
raise ValueError("账号需字母开头,3-32位,仅含字母/数字/_ . -")
return v
@model_validator(mode='before')
@classmethod
def _normalize(cls, values):
if isinstance(values, dict):
for k in ["name", "username", "password", "description"]:
if k in values and isinstance(values[k], str):
values[k] = values[k].strip() or values[k]
# role_ids 去重并转为 int
if "role_ids" in values and values["role_ids"] is not None:
try:
values["role_ids"] = list[int]({int(x) for x in values["role_ids"]})
except Exception:
pass
# mobile 空串转 None
if "mobile" in values and isinstance(values["mobile"], str) and values["mobile"].strip() == "":
values["mobile"] = None
return values
class UserForgetPasswordSchema(BaseModel):
"""忘记密码"""
@@ -116,35 +98,6 @@ class UserCreateSchema(CurrentUserUpdateSchema):
role_ids: Optional[List[int]] = Field(default=[], description='角色ID')
position_ids: Optional[List[int]] = Field(default=[], description='岗位ID')
@model_validator(mode='before')
@classmethod
def _normalize(cls, values):
if isinstance(values, dict):
# 字符串去空格和空串转 None
for k in ["username", "password", "description", "name"]:
if k in values and isinstance(values[k], str):
values[k] = values[k].strip() or None if values[k].strip() == "" else values[k].strip()
# bool 兼容
for k in ["status", "is_superuser"]:
if k in values:
v = values[k]
if isinstance(v, str):
values[k] = v.strip().lower() in {"true", "1", "yes", "y"}
# 列表转 int 去重
for k in ["role_ids", "position_ids"]:
if k in values and values[k] is not None:
try:
values[k] = list({int(x) for x in values[k]})
except Exception:
pass
return values
@model_validator(mode='after')
def _validate_after(self):
if self.status is False and (not self.description or not str(self.description).strip()):
raise ValueError("禁用状态下必须填写备注描述")
return self
class UserUpdateSchema(UserCreateSchema):
"""更新"""