c83287819a
- 操作日志审计机制：audit 上下文（成功与失败都记、独立提交、操作人标量 固化、commit 后查询前置约定），用户/角色全部写操作接入；request_data 按键名的结构化脱敏 + UTF-8 字节边界截断 - 角色管理：CRUD 与启停；admin 禁删禁停、有绑定禁删（FOR UPDATE 锁行）、 编码创建后不可改、停用角色不可再分配 - 日志查询：操作/登录两端点，列表与详情两段式（详情含已脱敏请求参数）； 时间参数统一转本地 naive（修混用时区 500 与偏移丢弃） - 用户管理：补齐 reset_password 操作人、编辑空串归一 NULL、并发唯一键兜底 400 - 测试 301 例全绿（单元 223 / 集成 78） Co-Authored-By: Claude Code <noreply@anthropic.com>
114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
"""用户管理的请求/响应模型（角色相关模型见 `role.py`）。
|
||
|
||
与 `auth.py` 的分工：`auth.py` 管登录链路（验证码、登录、当前用户），
|
||
本模块管**用户管理**（增删改查、重置密码、启停、角色分配）。
|
||
|
||
字段名驼峰（规范约定）与蛇形属性之间用 alias 映射，与响应模型保持同一套做法。
|
||
"""
|
||
|
||
from datetime import datetime
|
||
|
||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||
|
||
from app.core.constants import PASSWORD_MAX_BYTES, PASSWORD_MIN_BYTES
|
||
from app.core.response import PageQuery
|
||
|
||
|
||
def _validate_password_bytes(value: str) -> str:
|
||
"""按**字节数**校验密码长度。
|
||
|
||
必须在这里拦下：bcrypt 5.0 起对超过 72 字节的密码直接抛 ValueError，
|
||
而中文一个字符占 3 字节 —— 按字符数校验（如 max_length=20）根本拦不住。
|
||
与 `auth.py` 的登录校验共用同一规则，避免两处口径不一致。
|
||
"""
|
||
length = len(value.encode("utf-8"))
|
||
if length < PASSWORD_MIN_BYTES:
|
||
raise ValueError(f"密码不足 {PASSWORD_MIN_BYTES} 字节")
|
||
if length > PASSWORD_MAX_BYTES:
|
||
raise ValueError(f"密码超过 {PASSWORD_MAX_BYTES} 字节上限")
|
||
return value
|
||
|
||
|
||
class UserOut(BaseModel):
|
||
"""用户列表/详情响应。
|
||
|
||
不含 password —— 密码哈希绝不外泄，这是硬约束而非疏漏。
|
||
"""
|
||
|
||
model_config = ConfigDict(from_attributes=True)
|
||
|
||
id: int
|
||
username: str
|
||
real_name: str | None = None
|
||
phone: str | None = None
|
||
email: str | None = None
|
||
status: int = Field(description="状态：1启用 0禁用")
|
||
roles: list[str] = Field(default_factory=list, description="角色编码列表")
|
||
last_login_time: datetime | None = None
|
||
remark: str | None = None
|
||
create_time: datetime | None = None
|
||
|
||
|
||
class UserCreateIn(BaseModel):
|
||
"""新增用户。"""
|
||
|
||
model_config = ConfigDict(str_strip_whitespace=True)
|
||
|
||
username: str = Field(min_length=2, max_length=64, description="登录用户名，须唯一")
|
||
password: str = Field(description="初始密码")
|
||
real_name: str | None = Field(default=None, max_length=64)
|
||
phone: str | None = Field(default=None, max_length=20)
|
||
email: str | None = Field(default=None, max_length=128)
|
||
roles: list[str] = Field(default_factory=list, description="角色编码列表，至少一个")
|
||
remark: str | None = Field(default=None, max_length=255)
|
||
|
||
@field_validator("password")
|
||
@classmethod
|
||
def _check_password(cls, value: str) -> str:
|
||
return _validate_password_bytes(value)
|
||
|
||
|
||
class UserUpdateIn(BaseModel):
|
||
"""编辑用户。
|
||
|
||
**不含 username** —— 用户名是登录凭据与审计追溯的锚点，改名会让历史
|
||
操作日志与登录日志对不上。确需改名应走专门的流程（本期不提供）。
|
||
"""
|
||
|
||
model_config = ConfigDict(str_strip_whitespace=True)
|
||
|
||
real_name: str | None = Field(default=None, max_length=64)
|
||
phone: str | None = Field(default=None, max_length=20)
|
||
email: str | None = Field(default=None, max_length=128)
|
||
roles: list[str] | None = Field(default=None, description="传 null 表示不改角色")
|
||
remark: str | None = Field(default=None, max_length=255)
|
||
|
||
|
||
class PasswordResetIn(BaseModel):
|
||
"""重置密码（管理员操作，不需要原密码）。"""
|
||
|
||
new_password: str = Field(description="新密码")
|
||
|
||
@field_validator("new_password")
|
||
@classmethod
|
||
def _check_password(cls, value: str) -> str:
|
||
return _validate_password_bytes(value)
|
||
|
||
|
||
class UserStatusIn(BaseModel):
|
||
"""启用/禁用用户。"""
|
||
|
||
status: int = Field(ge=0, le=1, description="1启用 0禁用")
|
||
|
||
|
||
class UserQuery(PageQuery):
|
||
"""用户列表查询条件。
|
||
|
||
继承分页基类，额外支持按用户名/姓名模糊、按状态与角色筛选。
|
||
"""
|
||
|
||
username: str | None = Field(default=None, max_length=64, description="用户名，模糊匹配")
|
||
real_name: str | None = Field(default=None, max_length=64, description="姓名，模糊匹配")
|
||
status: int | None = Field(default=None, ge=0, le=1, description="状态筛选")
|
||
role_code: str | None = Field(default=None, max_length=32, description="角色编码筛选")
|