5ece9f6dab
仓储层的 list_*/count_* 此前以 keyword/status/offset/limit 四五个散参数 传参，调用点重复铺开、易漏传且不可扩展；按项目「参数 >4 封装 DTO」的 约定统一收敛为各模块的 Query 对象（如 RoleQuery、BusinessQuery）， 服务层与测试调用同步调整，行为不变。 - repositories 层查询方法签名改为 (session, query)，分页与筛选条件 由 Query 承载，_apply_filters 同步收口 - schemas 层补齐各模块 Query 模型（approval/business_role/report 等） - services 层调用点同步改为传 Query - tests 随之调整；新增用例：待办角色流转（财务通过后领导出现待办）、 参数预览顺序稳定 验证：全量 pytest 900 passed（11:08）；单元 526 passed；ruff check 绿。 Co-Authored-By: Claude Code <noreply@anthropic.com>
118 lines
4.2 KiB
Python
118 lines
4.2 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
|
||
from app.schemas.role import RoleQuery
|
||
|
||
# ==================== 读取 ====================
|
||
|
||
|
||
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, query: RoleQuery) -> list[Role]:
|
||
"""分页查询角色（管理列表：含停用角色）。
|
||
|
||
排序与下拉口径一致（sort_order, id），保持同一套相对次序。
|
||
"""
|
||
stmt = _apply_filters(select(Role), query)
|
||
stmt = stmt.order_by(Role.sort_order, Role.id).offset(query.offset).limit(query.page_size)
|
||
return list(session.execute(stmt).scalars())
|
||
|
||
|
||
def count_roles(session: Session, query: RoleQuery) -> int:
|
||
"""统计符合条件的角色总数，供分页使用。"""
|
||
stmt = _apply_filters(select(func.count(Role.id)), query)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def _apply_filters(stmt: Select, query: RoleQuery) -> Select:
|
||
"""应用通用筛选条件。列表与计数共用，避免两处口径不一致。
|
||
|
||
关键词匹配走 `contains(autoescape=True)`（转义 LIKE 通配符）——
|
||
角色编码普遍含下划线，直接拼接会让 `_` 当单字符通配执行。
|
||
"""
|
||
if query.keyword:
|
||
stmt = stmt.where(
|
||
or_(
|
||
Role.role_code.contains(query.keyword, autoescape=True),
|
||
Role.role_name.contains(query.keyword, autoescape=True),
|
||
)
|
||
)
|
||
if query.status is not None:
|
||
stmt = stmt.where(Role.status == query.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()
|