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>
124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
"""角色数据访问。
|
||
|
||
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
||
|
||
`get_all_roles` 与 `get_roles_by_codes` 原位于 `user_repository`，
|
||
因角色管理成为独立模块而迁入本文件。
|
||
"""
|
||
|
||
from sqlalchemy import Select, func, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Role, UserRole
|
||
|
||
# ==================== 读取 ====================
|
||
|
||
|
||
def get_by_id(session: Session, role_id: int, *, for_update: bool = False) -> Role | None:
|
||
"""按主键查询角色。不存在返回 None。
|
||
|
||
``for_update=True`` 走 ``SELECT ... FOR UPDATE`` 锁定角色行 —— 删除前的
|
||
「有用户绑定禁删」校验必须用它：校验与 DELETE 之间若发生并发绑定，
|
||
`user_roles` 的 CASCADE 会把新绑定一并静默剥离。锁行后，并发绑定的
|
||
外键检查会等待，删除得以基于最新状态做出（对方失败而非数据被吞）。
|
||
"""
|
||
if for_update:
|
||
stmt = select(Role).where(Role.id == role_id).with_for_update()
|
||
return session.execute(stmt).scalar_one_or_none()
|
||
return session.get(Role, role_id)
|
||
|
||
|
||
def get_by_code(session: Session, role_code: str) -> Role | None:
|
||
"""按编码查询角色。不存在返回 None。"""
|
||
return session.execute(select(Role).where(Role.role_code == role_code)).scalar_one_or_none()
|
||
|
||
|
||
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 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())
|
||
|
||
|
||
def list_roles(
|
||
session: Session,
|
||
*,
|
||
offset: int,
|
||
limit: int,
|
||
keyword: str | None = None,
|
||
status: int | None = None,
|
||
) -> list[Role]:
|
||
"""分页查询角色（管理列表：含停用角色）。
|
||
|
||
排序与下拉口径一致（sort_order, id），保持同一套相对次序。
|
||
"""
|
||
stmt = _apply_filters(select(Role), keyword=keyword, status=status)
|
||
stmt = stmt.order_by(Role.sort_order, Role.id).offset(offset).limit(limit)
|
||
return list(session.execute(stmt).scalars())
|
||
|
||
|
||
def count_roles(session: Session, *, keyword: str | None = None, status: int | None = None) -> int:
|
||
"""统计符合条件的角色总数，供分页使用。"""
|
||
stmt = _apply_filters(select(func.count(Role.id)), keyword=keyword, status=status)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def _apply_filters(stmt: Select, *, keyword: str | None, status: int | None) -> Select:
|
||
"""应用通用筛选条件。列表与计数共用，避免两处口径不一致。
|
||
|
||
关键词匹配走 `contains(autoescape=True)`（转义 LIKE 通配符）——
|
||
角色编码普遍含下划线，直接拼接会让 `_` 当单字符通配执行。
|
||
"""
|
||
if keyword:
|
||
stmt = stmt.where(
|
||
or_(
|
||
Role.role_code.contains(keyword, autoescape=True),
|
||
Role.role_name.contains(keyword, autoescape=True),
|
||
)
|
||
)
|
||
if status is not None:
|
||
stmt = stmt.where(Role.status == status)
|
||
return stmt
|
||
|
||
|
||
def count_users_by_role(session: Session, role_ids: list[int]) -> dict[int, int]:
|
||
"""批量统计各角色关联的用户数。
|
||
|
||
一次 GROUP BY 取回全部计数，而非在列表循环里逐行查询（N+1）。
|
||
"""
|
||
if not role_ids:
|
||
return {}
|
||
stmt = (
|
||
select(UserRole.role_id, func.count())
|
||
.where(UserRole.role_id.in_(role_ids))
|
||
.group_by(UserRole.role_id)
|
||
)
|
||
return {int(role_id): int(count) for role_id, count in session.execute(stmt)}
|
||
|
||
|
||
# ==================== 写入 ====================
|
||
|
||
|
||
def create(session: Session, role: Role) -> Role:
|
||
"""新增角色。只 flush 以取得自增 id。"""
|
||
session.add(role)
|
||
session.flush()
|
||
return role
|
||
|
||
|
||
def delete(session: Session, role: Role) -> None:
|
||
"""删除角色。
|
||
|
||
调用方（service）必须先校验该角色无用户绑定 —— `user_roles` 外键是
|
||
CASCADE，直接删会静默剥离绑定用户的权限。
|
||
"""
|
||
session.delete(role)
|
||
session.flush()
|