e22c4bcdc9
需求 2.5.1 用户账号管理，全部要求管理员权限（require_admin 依赖）：
- 接口：列表/详情/新增/编辑/删除/重置密码/启停/角色下拉，
角色下拉放 /users/meta/roles 以避开 /{user_id} 的路由匹配冲突
- 删除、禁用、改角色三处自我保护：不允许对当前登录账号执行，
避免管理员误操作后立刻失去全部管理权限且无法自行恢复
- 重置密码与禁用均提升 token_version，使已签发令牌立即失效，
不依赖 Redis 黑名单
- 密码按字节数校验（bcrypt 5.0 硬上限 72 字节，中文一字 3 字节）
- 用户列表按 id 倒序：create_time 为秒级精度，同秒数据分页结果不确定
- 修复 replace_roles 未同步关系集合的缺陷：改用 user.user_roles
集合追加，此前 session.add(UserRole(...)) 会表现为保存成功但返回空角色
- update_user 的自我保护原先依赖返回 None 的占位函数，恒不生效，
改为显式传入 operator_id
Co-Authored-By: Claude Code <noreply@anthropic.com>
162 lines
5.5 KiB
Python
162 lines
5.5 KiB
Python
"""用户与角色数据访问。
|
||
|
||
本层只 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
|
||
|
||
|
||
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:
|
||
"""应用通用筛选条件。列表与计数共用，避免两处口径不一致。"""
|
||
if username:
|
||
stmt = stmt.where(User.username.like(f"%{username}%"))
|
||
if real_name:
|
||
stmt = stmt.where(User.real_name.like(f"%{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 get_roles_by_codes(session: Session, role_codes: list[str]) -> list[Role]:
|
||
"""按编码批量取角色。用于校验前端传来的角色是否存在。"""
|
||
if not role_codes:
|
||
return []
|
||
stmt = select(Role).where(Role.role_code.in_(role_codes))
|
||
return list(session.execute(stmt).scalars())
|
||
|
||
|
||
def replace_roles(session: Session, user: User, roles: list[Role]) -> None:
|
||
"""整体替换用户的角色。
|
||
|
||
采用「先清后建」而非逐条比对：角色数量极少（本期仅两个），
|
||
差异计算的复杂度换不来收益，且整体替换的语义更直白。
|
||
|
||
**必须经 `user.user_roles` 集合追加，不能用 `session.add(UserRole(...))`**：
|
||
后者只把对象放进 session，不会同步到关系的已加载集合，于是同一事务内
|
||
再读 `user.user_roles` 仍是空的 —— 表现为「保存成功，但接口返回的角色
|
||
列表是 []」。经集合操作时 `user_id` 由关系自动填充，集合也立即反映新状态。
|
||
"""
|
||
user.user_roles.clear()
|
||
for role in roles:
|
||
user.user_roles.append(UserRole(role_id=role.id))
|
||
session.flush()
|
||
|
||
|
||
def get_all_roles(session: Session) -> list[Role]:
|
||
"""取全部启用角色，供前端下拉选择。"""
|
||
stmt = select(Role).where(Role.status == 1).order_by(Role.sort_order, Role.id)
|
||
return list(session.execute(stmt).scalars())
|