schema.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131
  1. import re
  2. from datetime import datetime
  3. from typing import Optional
  4. from pydantic import BaseModel, ConfigDict, Field, model_validator, field_validator
  5. from sqlalchemy.ext.asyncio import AsyncSession
  6. from app.api.v1.module_system.user.model import UserModel
  7. class SmsCodeSchema(BaseModel):
  8. # 验证手机号格式
  9. mobile: str = Field(..., description="手机号")
  10. # 短信模版
  11. template_code: str = Field(..., description="短信模版")
  12. # 验证码
  13. code: Optional[str] = Field(default=None, description="验证码")
  14. @field_validator("mobile", mode="before")
  15. @classmethod
  16. def validate_mobile(cls, v: str) -> str:
  17. # 移除常见地格式化字符
  18. cleaned = re.sub(r'[\s\-+()()]', '', str(v))
  19. # 中国大陆手机号验证
  20. china_pattern = r'^1[3-9]\d{9}$'
  21. # 带国际区号的手机号
  22. # china_with_country_code = r'^(?:\+?86)?1[3-9]\d{9}$'
  23. # if not re.match(china_with_country_code, cleaned):
  24. # raise ValueError("手机号格式错误")
  25. if not re.match(china_pattern, cleaned):
  26. raise ValueError("手机号格式错误")
  27. # 返回清洗后的纯数字
  28. return cleaned.lstrip('+86')
  29. class AuthSchema(BaseModel):
  30. """权限认证模型"""
  31. model_config = ConfigDict(arbitrary_types_allowed=True)
  32. user: UserModel | None = Field(default=None, description="用户信息")
  33. check_data_scope: bool = Field(default=True, description="是否检查数据权限")
  34. db: AsyncSession = Field(description="数据库会话")
  35. tenant_id: int = Field(default=-1, description="租户ID")
  36. class JWTPayloadSchema(BaseModel):
  37. """JWT载荷模型"""
  38. sub: str = Field(..., description="用户登录信息")
  39. is_refresh: bool = Field(default=False, description="是否刷新token")
  40. exp: datetime | int = Field(..., description="过期时间")
  41. @model_validator(mode="after")
  42. def validate_fields(self):
  43. """
  44. 校验 JWT 载荷字段的基本合法性。
  45. 返回:
  46. - JWTPayloadSchema: 校验后的载荷实例。
  47. 异常:
  48. - ValueError: 必填字段为空或格式不正确时抛出。
  49. """
  50. if not self.sub or len(self.sub.strip()) == 0:
  51. raise ValueError("会话编号不能为空")
  52. return self
  53. class JWTOutSchema(BaseModel):
  54. """JWT响应模型"""
  55. model_config = ConfigDict(from_attributes=True)
  56. access_token: str = Field(..., min_length=1, description="访问token")
  57. refresh_token: str = Field(..., min_length=1, description="刷新token")
  58. token_type: str = Field(default="Bearer", description="token类型")
  59. expires_in: int = Field(..., gt=0, description="过期时间(秒)")
  60. class RefreshTokenPayloadSchema(BaseModel):
  61. """刷新Token载荷模型"""
  62. refresh_token: str = Field(..., min_length=1, description="刷新token")
  63. class LogoutPayloadSchema(BaseModel):
  64. """退出登录载荷模型"""
  65. token: str = Field(..., min_length=1, description="token")
  66. class CaptchaOutSchema(BaseModel):
  67. """验证码响应模型"""
  68. model_config = ConfigDict(from_attributes=True)
  69. enable: bool = Field(default=True, description="是否启用验证码")
  70. key: str = Field(..., min_length=1, description="验证码唯一标识")
  71. img_base: str = Field(..., min_length=1, description="Base64编码的验证码图片")
  72. class AutoLoginUserSchema(BaseModel):
  73. """免登录用户信息模型"""
  74. model_config = ConfigDict(from_attributes=True)
  75. id: int = Field(..., description="用户ID")
  76. username: str = Field(..., description="用户名")
  77. name: str = Field(..., description="用户姓名")
  78. avatar: str | None = Field(default=None, description="头像")
  79. class AutoLoginTokenSchema(BaseModel):
  80. """免登录Token响应模型"""
  81. model_config = ConfigDict(from_attributes=True)
  82. token: str = Field(..., description="免登录Token")
  83. user: AutoLoginUserSchema = Field(..., description="用户信息")
  84. class LoginMiniRequestSchema(BaseModel):
  85. """小程序登录请求模型"""
  86. model_config = ConfigDict(from_attributes=True)
  87. username: str = Field(..., min_length=1, description="用户名")
  88. password: str = Field(..., min_length=1, description="密码")
  89. login_type: Optional[str] = Field(default="mini", description="登录类型")