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>
159 lines
5.5 KiB
Python
159 lines
5.5 KiB
Python
"""用户数据访问（角色相关的读取与维护见 `role_repository`）。
|
||
|
||
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
||
"""
|
||
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import Select, func, select
|
||
from sqlalchemy.orm import Session, selectinload
|
||
|
||
from app.models import Role, User, UserRole
|
||
from app.repositories.query_utils import fuzzy
|
||
|
||
|
||
def _base_query() -> Select[tuple[User]]:
|
||
"""用户查询基座。
|
||
|
||
预加载角色关联，避免列表渲染时逐行查角色（N+1）。
|
||
"""
|
||
return select(User).options(selectinload(User.user_roles).selectinload(UserRole.role))
|
||
|
||
|
||
def get_by_username(session: Session, username: str) -> User | None:
|
||
"""按用户名查询用户。不存在返回 None。"""
|
||
return session.execute(select(User).where(User.username == username)).scalar_one_or_none()
|
||
|
||
|
||
def get_by_id(session: Session, user_id: int) -> User | None:
|
||
"""按主键查询用户。不存在返回 None。"""
|
||
return session.get(User, user_id)
|
||
|
||
|
||
def get_role_codes(user: User) -> list[str]:
|
||
"""取用户的角色编码列表。
|
||
|
||
依赖 `User.user_roles` 的 `lazy="selectin"` 预加载（见模型定义），
|
||
故此处遍历不会触发 N+1 查询。
|
||
"""
|
||
return [link.role.role_code for link in user.user_roles if link.role is not None]
|
||
|
||
|
||
def touch_last_login(session: Session, user: User, moment: datetime) -> None:
|
||
"""更新最后登录时间。只 flush，提交由 service 决定。"""
|
||
user.last_login_time = moment
|
||
session.flush()
|
||
|
||
|
||
# ==================== 列表查询 ====================
|
||
|
||
|
||
def list_users(
|
||
session: Session,
|
||
*,
|
||
offset: int,
|
||
limit: int,
|
||
username: str | None = None,
|
||
real_name: str | None = None,
|
||
status: int | None = None,
|
||
role_code: str | None = None,
|
||
) -> list[User]:
|
||
"""分页查询用户列表。
|
||
|
||
**按 id 倒序而非 create_time**：create_time 是秒级精度，批量导入的用户
|
||
可能落在同一秒，按它排序的分页结果不确定 —— 翻页时会出现重复或遗漏。
|
||
id 是自增主键，天然唯一且与创建顺序一致。
|
||
"""
|
||
stmt = _base_query()
|
||
stmt = _apply_filters(stmt, username=username, real_name=real_name, status=status)
|
||
if role_code:
|
||
# 按角色筛选需 join 关联表；角色是字典数据，数量极少，join 代价可忽略
|
||
stmt = stmt.join(UserRole).join(Role).where(Role.role_code == role_code)
|
||
stmt = stmt.order_by(User.id.desc()).offset(offset).limit(limit)
|
||
return list(session.execute(stmt).scalars().unique())
|
||
|
||
|
||
def count_users(
|
||
session: Session,
|
||
*,
|
||
username: str | None = None,
|
||
real_name: str | None = None,
|
||
status: int | None = None,
|
||
role_code: str | None = None,
|
||
) -> int:
|
||
"""统计符合条件的用户总数，供分页使用。"""
|
||
stmt = select(func.count(User.id))
|
||
stmt = _apply_filters(stmt, username=username, real_name=real_name, status=status)
|
||
if role_code:
|
||
stmt = stmt.join(UserRole).join(Role).where(Role.role_code == role_code)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def _apply_filters(
|
||
stmt: Select,
|
||
*,
|
||
username: str | None,
|
||
real_name: str | None,
|
||
status: int | None,
|
||
) -> Select:
|
||
"""应用通用筛选条件。列表与计数共用，避免两处口径不一致。
|
||
|
||
模糊匹配统一走 `query_utils.fuzzy`（转义 LIKE 通配符）—— 用户名普遍
|
||
含下划线，直接拼接会让 `_` 当单字符通配执行、误配无关账号。
|
||
"""
|
||
if username:
|
||
stmt = fuzzy(stmt, User.username, username)
|
||
if real_name:
|
||
stmt = fuzzy(stmt, User.real_name, real_name)
|
||
if status is not None:
|
||
stmt = stmt.where(User.status == status)
|
||
return stmt
|
||
|
||
|
||
# ==================== 写入 ====================
|
||
|
||
|
||
def create(session: Session, user: User) -> User:
|
||
"""新增用户。只 flush 以取得自增 id。"""
|
||
session.add(user)
|
||
session.flush()
|
||
return user
|
||
|
||
|
||
def delete(session: Session, user: User) -> None:
|
||
"""物理删除用户。
|
||
|
||
本项目对用户采用「删除即禁用」的**业务**约定（见模型注释），
|
||
但保留物理删除能力供测试清理与误建账号的彻底移除使用。
|
||
`user_roles` 与 `operation_logs` 的外键分别配置了 CASCADE 与 SET NULL，
|
||
故不会留下悬挂引用。
|
||
"""
|
||
session.delete(user)
|
||
session.flush()
|
||
|
||
|
||
# ==================== 角色关联 ====================
|
||
|
||
|
||
def replace_roles(session: Session, user: User, roles: list[Role]) -> None:
|
||
"""整体替换用户的角色。
|
||
|
||
采用「先清后建」而非逐条比对：角色数量极少（本期仅两个），
|
||
差异计算的复杂度换不来收益，且整体替换的语义更直白。
|
||
|
||
**clear 之后必须立刻 flush 一次**：SQLAlchemy 在同一事务的 flush 中
|
||
**先处理 INSERT 再处理 DELETE**，若中间不落库，重插与旧集合有交集的
|
||
(user_id, role_id)（编辑用户时保留既有角色就会发生）会撞
|
||
`uk_user_roles_user_id_role_id` 唯一键，报 1062。
|
||
|
||
**必须经 `user.user_roles` 集合追加，不能用 `session.add(UserRole(...))`**：
|
||
后者只把对象放进 session，不会同步到关系的已加载集合，于是同一事务内
|
||
再读 `user.user_roles` 仍是空的 —— 表现为「保存成功，但接口返回的角色
|
||
列表是 []」。经集合操作时 `user_id` 由关系自动填充，集合也立即反映新状态。
|
||
"""
|
||
user.user_roles.clear()
|
||
session.flush()
|
||
for role in roles:
|
||
user.user_roles.append(UserRole(role_id=role.id))
|
||
session.flush()
|