Add the schema base class (#148)

2026-09-21 21:15:13 +00:00 · 2023-06-18 19:47:39 +08:00
parent 616243c0dc
commit 59f1daba90
16 changed files with 150 additions and 61 deletions
@@ -5,6 +5,7 @@ import json
 from fastapi import FastAPI, Request
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
+from pydantic.errors import EnumMemberError, WrongConstantError
 from starlette.exceptions import HTTPException
 from starlette.responses import JSONResponse
 from uvicorn.protocols.http.h11_impl import STATUS_PHRASES
@@ -66,22 +67,30 @@ def register_exception(app: FastAPI):
        message = ''
        data = {}
        for raw_error in exc.raw_errors:
-            if isinstance(raw_error.exc, ValidationError):
-                exc = raw_error.exc
-                if hasattr(exc, 'model'):
-                    fields = exc.model.__dict__.get('__fields__')
+            raw_exc = raw_error.exc
+            if isinstance(raw_exc, ValidationError):
+                if hasattr(raw_exc, 'model'):
+                    fields = raw_exc.model.__dict__.get('__fields__')
                    for field_key in fields.keys():
                        field_title = fields.get(field_key).field_info.title
                        data[field_key] = field_title if field_title else field_key
-                errors_len = len(exc.errors())
-                for error in exc.errors():
+                # 处理特殊类型异常模板信息 -> backend/app/schemas/base.py: SCHEMA_ERROR_MSG_TEMPLATES
+                sub_raw_exc = raw_exc.raw_errors[0].exc
+                if isinstance(sub_raw_exc, (EnumMemberError, WrongConstantError)):
+                    if getattr(sub_raw_exc, 'code') == 'enum':
+                        sub_raw_exc.__dict__['permitted'] = ', '.join(repr(v.value) for v in sub_raw_exc.enum_values)  # type: ignore  # noqa: E501
+                    else:
+                        sub_raw_exc.__dict__['permitted'] = ', '.join(repr(v) for v in sub_raw_exc.permitted)  # type: ignore  # noqa: E501
+                # 处理异常信息
+                errors_len = len(raw_exc.errors())
+                for error in raw_exc.errors():
                    field = str(error.get('loc')[-1])
-                    _msg = error.get('msg')
+                    msg = error.get('msg')
                    errors_len = errors_len - 1
                    message += (
-                        f'{data.get(field, field) if field != "__root__" else ""} {_msg}' + ', '
+                        f'{data.get(field, field) if field != "__root__" else ""} {msg}' + ', '
                        if errors_len > 0
-                        else f'{data.get(field, field) if field != "__root__" else ""} {_msg}' + '.'
+                        else f'{data.get(field, field) if field != "__root__" else ""} {msg}' + '.'
                    )
            elif isinstance(raw_error.exc, json.JSONDecodeError):
                message += 'json解析失败'
@@ -22,10 +22,10 @@ class Dept(Base):
    email: Mapped[str | None] = mapped_column(String(50), default=None, comment='邮箱')
    status: Mapped[int] = mapped_column(default=1, comment='部门状态(0停用 1正常)')
    del_flag: Mapped[bool] = mapped_column(default=False, comment='删除标志（0删除 1存在）')
+    # 父级部门一对多
    parent_id: Mapped[int | None] = mapped_column(
        ForeignKey('sys_dept.id', ondelete='SET NULL'), default=None, index=True, comment='父部门ID'
    )
-    # 父级部门一对多
    parent: Mapped[Union['Dept', None]] = relationship(init=False, back_populates='children', remote_side=[id])
    children: Mapped[list['Dept'] | None] = relationship(init=False, back_populates='parent')
    # 部门用户一对多
@@ -15,9 +15,9 @@ class DictData(Base):
    id: Mapped[id_key] = mapped_column(init=False)
    label: Mapped[str] = mapped_column(String(32), unique=True, comment='字典标签')
    value: Mapped[str] = mapped_column(String(32), unique=True, comment='字典值')
-    type_id: Mapped[int] = mapped_column(ForeignKey('sys_dict_type.id'), comment='字典类型id')
    sort: Mapped[int] = mapped_column(default=0, comment='排序')
    status: Mapped[int] = mapped_column(default=1, comment='状态（0停用 1正常）')
    remark: Mapped[str | None] = mapped_column(LONGTEXT, default=None, comment='备注')
    # 字典类型一对多
+    type_id: Mapped[int] = mapped_column(ForeignKey('sys_dict_type.id'), default=None, comment='字典类型关联ID')
    type: Mapped['DictType'] = relationship(init=False, back_populates='datas')  # noqa: F821
@@ -26,10 +26,10 @@ class Menu(Base):
    perms: Mapped[str | None] = mapped_column(String(100), default=None, comment='权限标识')
    status: Mapped[int] = mapped_column(default=1, comment='菜单状态（0停用 1正常）')
    remark: Mapped[str | None] = mapped_column(LONGTEXT, default=None, comment='备注')
+    # 父级菜单一对多
    parent_id: Mapped[int | None] = mapped_column(
        ForeignKey('sys_menu.id', ondelete='SET NULL'), default=None, index=True, comment='父菜单ID'
    )
-    # 父级菜单一对多
    parent: Mapped[Union['Menu', None]] = relationship(init=False, back_populates='children', remote_side=[id])
    children: Mapped[list['Menu'] | None] = relationship(init=False, back_populates='parent')
    # 菜单角色多对多
@@ -2,14 +2,15 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel, Field, validator
+from pydantic import Field, validator

 from backend.app.common.enums import MethodType
+from backend.app.schemas.base import SchemaBase


-class ApiBase(BaseModel):
+class ApiBase(SchemaBase):
    name: str
-    method: str = Field(default=MethodType.GET, description='请求方法')
+    method: MethodType = Field(default=MethodType.GET, description='请求方法')
    path: str = Field(..., description='api路径')
    remark: str | None = None

@@ -17,9 +18,6 @@ class ApiBase(BaseModel):
    def method_validator(cls, v):
        if not v.isupper():
            raise ValueError('请求方式必须大写')
-        allow_method = MethodType.get_member_values()
-        if v not in allow_method:
-            raise ValueError(f'请求方式不合法, 仅支持: {allow_method}')
        return v


@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from pydantic import BaseModel
+
+SCHEMA_ERROR_MSG_TEMPLATES: dict[str, str] = {
+    # Type Errors
+    'type_error.arbitrary_type': '预期为 {expected_arbitrary_type} 的实例',
+    'type_error.bool': '值不是有效的布尔值',
+    'type_error.callable': '{value} 不可调用',
+    'type_error.class': '预计将是一个类',
+    'type_error.dataclass': '{class_name} 的实例，预期为元组或字典',
+    'type_error.decimal': '值不是有效的小数（Decimal）',
+    'type_error.deque': '值不是有效的双端队列',
+    'type_error.dict': '值不是有效的字典',
+    'type_error.enum_instance': '{value} 不是有效的枚举实例',
+    'type_error.enum': '值不是有效的枚举成员; 允许：{permitted}',
+    'type_error.float': '值不是有效的浮点数',
+    'type_error.frozenset': '值不是有效的冻结集',
+    'type_error.hashable': '值不是有效的哈希值',
+    'type_error.int_enum_instance': '{value} 不是有效的 IntEnum 实例',
+    'type_error.integer': '值不是有效的整数',
+    'type_error.iterable': '值不是有效的可迭代对象',
+    'type_error.json': 'JSON 对象必须是字符串，字节或字节数组',
+    'type_error.list': '值不是有效的列表',
+    'type_error.none.allowed': '值不是 None',
+    'type_error.none.not_allowed': '不允许的值: None',
+    'type_error.not_none': '值不是 None',
+    'type_error.path': '值不是有效路径',
+    'type_error.pyobject': '确保此值包含有效的导入路径或有效的可调用对象：{error_message}',
+    'type_error.sequence': '值不是有效的序列',
+    'type_error.set': '值不是有效的集合',
+    'type_error.subclass': '预期 {expected_class} 的子类',
+    'type_error.tuple': '值不是有效的元组',
+    'type_error.uuid': '值不是有效的 UUID',
+
+    # Value Errors
+    'value_error.any_str.max_length': '确保此值最多包含 {limit_value} 个字符',
+    'value_error.any_str.min_length': '确保此值至少包含 {limit_value} 个字符',
+    'value_error.color': '值不是有效的颜色: {reason}',
+    'value_error.const': '意外值; 允许: {permitted}',
+    'value_error.date.not_in_the_future': '日期不在未来时间',
+    'value_error.date.not_in_the_past': '日期不是过去时间',
+    'value_error.decimal.max_digits': '确保总共不超过 {max_digits} 位数字',
+    'value_error.decimal.max_places': '确保小数位数不超过 {decimal_places} 位',
+    'value_error.decimal.not_finite': '值不是有效的小数（Decimal）',
+    'value_error.decimal.whole_digits': '确保小数点前不超过 {whole_digits} 位',
+    'value_error.discriminated_union.invalid_discriminator': '不匹配鉴别器 {discriminator_key!r} 和值 {discriminator_value!r}（允许的值：{allowed_values}）',
+    'value_error.discriminated_union.missing_discriminator': '鉴别器 {discriminator_key!r} 的值缺失',
+    'value_error.extra': '不允许使用额外字段',
+    'value_error.frozenset.max_items': '确保此值最多包含 {limit_value} 个项目',
+    'value_error.frozenset.min_items': '确保此值至少包含 {limit_value} 个项目',
+    'value_error.invalidbytesizeunit': '无法解释字节单位: {unit}',
+    'value_error.list.max_items': '确保此值最多包含 {limit_value} 个项目',
+    'value_error.list.min_items': '确保此值至少包含 {limit_value} 个项目',
+    'value_error.list.unique_items': '列表包含重复项',
+    'value_error.missing': '必填字段',
+    'value_error.number.not_finite_number': '确保此值是有限数',
+    'value_error.number.not_ge': '确保此值大于或等于 {limit_value}',
+    'value_error.number.not_gt': '确保此值大于 {limit_value}',
+    'value_error.number.not_le': '确保此值小于或等于 {limit_value}',
+    'value_error.number.not_lt': '确保此值小于 {limit_value}。',
+    'value_error.number.not_multiple': '确保此值是 {multiple_of} 的倍数。',
+    'value_error.path.not_a_directory': '路径 "{path}" 没有指向一个目录',
+    'value_error.path.not_a_file': '路径 "{path}" 没有指向一个文件',
+    'value_error.path.not_exists': '路径为 "{path}" 的文件或目录不存在',
+    'value_error.payment_card_number.digits': '卡号不全是数字',
+    'value_error.payment_card_number.invalid_length_for_brand': '{brand} 卡的长度必须为 {required_length}',
+    'value_error.payment_card_number.luhn_check': '卡号无效',
+    'value_error.regex_pattern': '无效的正则表达式',
+    'value_error.set.max_items': '确保此值最多有 {limit_value} 项',
+    'value_error.set.min_items': '确保此值至少有 {limit_value} 项',
+    'value_error.str.regex': '字符串不符合正则 "{pattern}" 的要求。',
+    'value_error.tuple.length': '错误的元组长度 {actual_length}，预期的 {expected_length}。',
+    'value_error.url.extra': 'URL 无效，在有效的 URL 后发现额外的字符： [extra!r].',
+    'value_error.url.host': 'URL 地址无效',
+    'value_error.url.port': 'URL 端口无效，端口不能超过 65535',
+    'value_error.url.scheme': '不允许的 URL 方案',
+    'value_error.url.userinfo': 'URL 中需要用户信息，但缺少用户信息',
+    'value_error.uuid.version': 'UUID 预期的版本 {required_version}',
+}
+
+
+class SchemaBase(BaseModel):
+    class Config:
+        use_enum_values = True
+        # 错误信息模板对于模型嵌套无效
+        # https://github.com/pydantic/pydantic/issues/5651
+        error_msg_templates = SCHEMA_ERROR_MSG_TEMPLATES
@@ -1,22 +1,20 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-from pydantic import BaseModel, Field, validator
+from pydantic import Field, validator

 from backend.app.common.enums import MethodType
+from backend.app.schemas.base import SchemaBase


-class CreatePolicy(BaseModel):
+class CreatePolicy(SchemaBase):
    sub: str = Field(..., description='用户uuid / 角色')
    path: str = Field(..., description='api 路径')
-    method: str = Field(default=MethodType.GET, description='请求方法')
+    method: MethodType = Field(default=MethodType.GET, description='请求方法')

    @validator('method')
    def method_validator(cls, v):
        if not v.isupper():
            raise ValueError('请求方式必须大写')
-        allow_method = MethodType.get_member_values()
-        if v not in allow_method:
-            raise ValueError(f'请求方式不合法, 仅支持: {allow_method}')
        return v


@@ -28,7 +26,7 @@ class DeletePolicy(CreatePolicy):
    pass


-class CreateUserRole(BaseModel):
+class CreateUserRole(SchemaBase):
    uuid: str = Field(..., description='用户 uuid')
    role: str = Field(..., description='角色')

@@ -37,7 +35,7 @@ class DeleteUserRole(CreateUserRole):
    pass


-class GetAllPolicy(BaseModel):
+class GetAllPolicy(SchemaBase):
    id: int
    ptype: str = Field(..., description='规则类型, p 或 g')
    v0: str = Field(..., description='用户 uuid / 角色')
@@ -2,13 +2,14 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel, Field, validator
+from pydantic import Field, validator

 from backend.app.common.enums import StatusType
+from backend.app.schemas.base import SchemaBase
 from backend.app.utils.re_verify import is_phone


-class DeptBase(BaseModel):
+class DeptBase(SchemaBase):
    name: str
    parent_id: int | None = Field(default=None, ge=1, description='菜单父级ID')
    sort: int = Field(default=0, ge=0, description='排序')
@@ -2,19 +2,20 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel, Field
+from pydantic import Field

 from backend.app.common.enums import StatusType
+from backend.app.schemas.base import SchemaBase
 from backend.app.schemas.dict_type import GetAllDictType


-class DictDataBase(BaseModel):
+class DictDataBase(SchemaBase):
+    type_id: int
    label: str
    value: str
    sort: int
    status: StatusType = Field(default=StatusType.enable)
    remark: str | None = None
-    type_id: int


 class CreateDictData(DictDataBase):
@@ -2,12 +2,13 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel, Field
+from pydantic import Field

 from backend.app.common.enums import StatusType
+from backend.app.schemas.base import SchemaBase


-class DictTypeBase(BaseModel):
+class DictTypeBase(SchemaBase):
    name: str
    code: str
    status: StatusType = Field(default=StatusType.enable)
@@ -2,10 +2,11 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel
+
+from backend.app.schemas.base import SchemaBase


-class LoginLogBase(BaseModel):
+class LoginLogBase(SchemaBase):
    user_uuid: str
    username: str
    status: int
@@ -2,12 +2,13 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel, Field, validator
+from pydantic import Field

 from backend.app.common.enums import MenuType, StatusType
+from backend.app.schemas.base import SchemaBase


-class MenuBase(BaseModel):
+class MenuBase(SchemaBase):
    name: str
    parent_id: int = Field(default=None, ge=1, description='菜单父级ID')
    sort: int = Field(default=0, ge=0, description='排序')
@@ -19,12 +20,6 @@ class MenuBase(BaseModel):
    status: StatusType = Field(default=StatusType.enable)
    remark: str | None = None

-    @validator('menu_type')
-    def menu_type_validator(cls, v):
-        if v not in MenuType.get_member_values():
-            raise ValueError('菜单类型只能是 0，1，2')
-        return v
-

 class CreateMenu(MenuBase):
    pass
@@ -2,12 +2,13 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel, Field
+from pydantic import Field

 from backend.app.common.enums import StatusType
+from backend.app.schemas.base import SchemaBase


-class OperaLogBase(BaseModel):
+class OperaLogBase(SchemaBase):
    username: str | None = None
    method: str
    title: str
@@ -2,24 +2,19 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel, Field, validator
+from pydantic import Field

 from backend.app.common.enums import RoleDataScope, StatusType
+from backend.app.schemas.base import SchemaBase
 from backend.app.schemas.menu import GetAllMenu


-class RoleBase(BaseModel):
+class RoleBase(SchemaBase):
    name: str
    data_scope: RoleDataScope = Field(default=RoleDataScope.custom, description='数据范围（1：全部数据权限 2：自定数据权限）')  # noqa: E501
    status: StatusType = Field(default=StatusType.enable)
    remark: str | None = None

-    @validator('data_scope')
-    def data_scope_validator(cls, v):
-        if v not in RoleDataScope.get_member_values():
-            raise ValueError('数据范围只能是 1 或 2')
-        return v
-

 class CreateRole(RoleBase):
    menus: list[int]
@@ -2,18 +2,18 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime

-from pydantic import BaseModel

+from backend.app.schemas.base import SchemaBase
 from backend.app.schemas.user import GetUserInfoNoRelation


-class GetSwaggerToken(BaseModel):
+class GetSwaggerToken(SchemaBase):
    access_token: str
    token_type: str = 'Bearer'
    user: GetUserInfoNoRelation


-class AccessTokenBase(BaseModel):
+class AccessTokenBase(SchemaBase):
    access_token: str
    access_token_type: str = 'Bearer'
    access_token_expire_time: datetime
@@ -3,14 +3,15 @@
 from datetime import datetime

 from email_validator import validate_email, EmailNotValidError
-from pydantic import BaseModel, HttpUrl, Field, validator
+from pydantic import HttpUrl, Field, validator

 from backend.app.common.enums import StatusType
+from backend.app.schemas.base import SchemaBase
 from backend.app.schemas.dept import GetAllDept
 from backend.app.schemas.role import GetAllRole


-class Auth(BaseModel):
+class Auth(SchemaBase):
    username: str
    password: str

@@ -34,7 +35,7 @@ class CreateUser(Auth):
        return v


-class _UserInfoBase(BaseModel):
+class _UserInfoBase(SchemaBase):
    dept_id: int
    username: str
    nickname: str
@@ -60,7 +61,7 @@ class UpdateUser(_UserInfoBase):
    roles: list[int]


-class Avatar(BaseModel):
+class Avatar(SchemaBase):
    url: HttpUrl = Field(..., description='头像 http 地址')


@@ -87,7 +88,7 @@ class GetAllUserInfo(GetUserInfoNoRelation):
        orm_mode = True


-class ResetPassword(BaseModel):
+class ResetPassword(SchemaBase):
    old_password: str
    new_password: str
    confirm_password: str