iTi-Flask-IOT/iti/applications/routes/sys/schemas/user.py

from dataclasses import field
import datetime
from marshmallow_dataclass import dataclass
from iti.applications.common.enums import GenderEnum, StatusEnum
from marshmallow import validates_schema, ValidationError
from iti.applications.common.utils.schema import BaseSchema, Pagination
from enum import Enum
from typing import ClassVar, Optional
from apiflask import fields


@dataclass(base_schema=BaseSchema)
class UserQuery(Pagination):
    """
    用户查询请求
    """

    keyword: Optional[str] = field(
        default=None,
        metadata={
            "required": False,
            "metadata": {
                "description": "关键字 [用户名|手机号|邮箱|真实姓名] 模糊查询"
            },
        },
    )
    username: str = field(
        default=None,
        metadata={
            "required": False,
            "metadata": {"example": "admin", "description": "用户名"},
        },
    )
    phone: str = field(
        default=None,
        metadata={
            "required": False,
            "metadata": {"example": "18888888888", "description": "手机号"},
        },
    )
    email: str = field(
        default=None,
        metadata={
            "required": False,
            "metadata": {"example": "admin@example.com", "description": "邮箱"},
        },
    )
    realname: str = field(
        default=None,
        metadata={
            "required": False,
            "metadata": {"example": "张三", "description": "真实姓名"},
        },
    )
    gender: GenderEnum = field(
        default=None,
        metadata={
            "required": False,
            "by_value": True,
            "metadata": {
                "example": "male",
                "description": "性别",
            },
        },
    )
    status: StatusEnum = field(
        default=None,
        metadata={
            "required": False,
            "by_value": True,
            "metadata": {"example": "enabled", "description": "状态"},
        },
    )
    createdAt: list[datetime.datetime] = field(
        default=None,
        metadata={
            "required": False,
            "data_key": "createdAt[]",
            "metadata": {
                "description": "创建时间范围",
                "example": [
                    "2025-01-01 00:00:00",
                    "2025-01-01 23:59:59",
                ],
            },
        },
    )
    updatedAt: list[datetime.datetime] = field(
        default=None,
        metadata={
            "required": False,
            "data_key": "updatedAt[]",
            "metadata": {
                "example": [
                    "2025-01-01 00:00:00",
                    "2025-01-01 23:59:59",
                ],
                "description": "更新时间范围",
            },
        },
    )
    Schema: ClassVar[BaseSchema] = BaseSchema  # For the type check


class UserCreateRequest(BaseSchema):
    """
    创建用户请求
    """

    username = fields.String(
        required=False,
        metadata={"example": "admin", "description": "用户名"},
        load_default=None,
    )
    phone = fields.String(
        required=False,
        metadata={"example": "18888888888", "description": "手机号"},
        load_default=None,
    )
    email = fields.String(
        required=False,
        metadata={"example": "admin@example.com", "description": "邮箱"},
        load_default=None,
    )
    password = fields.String(
        required=False,
        metadata={"example": "123456", "description": "密码"},
        load_default=None,
    )
    realname = fields.String(
        required=False,
        metadata={"example": "张三", "description": "真实姓名"},
        load_default=None,
    )
    avatar = fields.String(
        required=False,
        metadata={"example": "https://example.com/avatar.jpg", "description": "头像"},
        load_default=None,
    )
    gender = fields.Enum(
        GenderEnum,
        by_value=True,
        required=False,
        metadata={"example": "secure", "description": "性别"},
        load_default=GenderEnum.SECURE,
    )
    status = fields.Enum(
        StatusEnum,
        by_value=True,
        required=False,
        metadata={"example": "enabled", "description": "状态"},
        load_default=StatusEnum.ENABLED,
    )
    roles = fields.List(
        fields.String(),
        required=False,
        metadata={"example": ["1", "2"], "description": "角色ID列表"},
        load_default=None,
    )
    depts = fields.List(
        fields.String(),
        required=False,
        metadata={"example": ["1", "2"], "description": "部门ID列表"},
        load_default=None,
    )

    @validates_schema
    def validate_schema(self, data, **kwargs):
        if not data.get("username") and not data.get("phone") and not data.get("email"):
            raise ValidationError(
                "至少需要提供一种主体（用户名、手机号、邮箱）", field_name="custom"
            )
        return data


class UserUpdateRequest(BaseSchema):
    """
    更新用户请求
    """

    username = fields.String(
        required=False,
        metadata={"example": "admin", "description": "用户名"},
        load_default=None,
    )
    phone = fields.String(
        required=False,
        metadata={"example": "18888888888", "description": "手机号"},
        load_default=None,
    )
    email = fields.String(
        required=False,
        metadata={"example": "admin@example.com", "description": "邮箱"},
        load_default=None,
    )
    realname = fields.String(
        required=False,
        metadata={"example": "张三", "description": "真实姓名"},
        load_default=None,
    )
    avatar = fields.String(
        required=False,
        metadata={"example": "https://example.com/avatar.jpg", "description": "头像"},
        load_default=None,
    )
    gender = fields.Enum(
        GenderEnum,
        by_value=True,
        required=False,
        metadata={"example": "secure", "description": "性别"},
        load_default=None,
    )
    status = fields.Enum(
        StatusEnum,
        by_value=True,
        required=False,
        metadata={"example": "enabled", "description": "状态"},
        load_default=None,
    )
    desc = fields.String(
        required=False,
        metadata={"example": "描述", "description": "描述"},
        load_default=None,
    )
    roles = fields.List(
        fields.String(),
        required=False,
        metadata={"example": ["1", "2"], "description": "角色ID列表"},
        load_default=None,
    )
    depts = fields.List(
        fields.String(),
        required=False,
        metadata={"example": ["1", "2"], "description": "部门ID列表"},
        load_default=None,
    )


class UserPasswordRequestType(str, Enum):
    """
    密码请求类型
    """

    UPDATE_BY_OLD_PASSWORD = "updateByOldPassword"  # 通过旧密码修改
    UPDATE_BY_VERIFY_CODE = "updateByVerifyCode"  # 通过验证码修改
    RESET_PASSWORD = "resetPassword"  # 重置密码


@dataclass(base_schema=BaseSchema)
class UserUpdatePasswordRequest:
    """
    修改密码请求
    """

    type: UserPasswordRequestType = field(
        metadata={
            "required": True,
            "by_value": True,
            "metadata": {
                "example": "updateByOldPassword",
                "description": "密码请求类型",
            },
        },
        default=UserPasswordRequestType.UPDATE_BY_OLD_PASSWORD,
    )
    code: str = field(
        metadata={
            "required": False,
            "metadata": {"example": "123456", "description": "验证码"},
        },
        default=None,
    )
    old_password: str = field(
        metadata={
            "required": False,
            "metadata": {"example": "123456", "description": "旧密码"},
        },
        default=None,
    )
    new_password: str = field(
        metadata={
            "required": True,
            "metadata": {"example": "123456", "description": "新密码"},
        },
        default=None,
    )
    Schema: ClassVar[BaseSchema] = BaseSchema  # For the type check

    @validates_schema
    def validate_schema(self, data, **kwargs):
        reqType = data.get("type")
        if not data.get("new_password"):
            raise ValidationError("新密码不能为空", field_name="new_password")
        if reqType == UserPasswordRequestType.UPDATE_BY_OLD_PASSWORD:
            if not data.get("old_password"):
                raise ValidationError("旧密码不能为空", field_name="old_password")
        if reqType == UserPasswordRequestType.UPDATE_BY_VERIFY_CODE:
            if not data.get("code"):
                raise ValidationError("验证码不能为空", field_name="code")
        if reqType == UserPasswordRequestType.RESET_PASSWORD:
            if not data.get("code") and not data.get("new_password"):
                raise ValidationError("验证码或新密码不能同时为空", field_name="custom")
        return data