liuwenxin
1e53792ad6
打通 T+ 快照、业务、发货、回款和核算准备链路，保留来源差异并提供待补状态。 新增业务自动接入任务、跨平台读取服务托管、权限复核、来源关联和前端核对页面。 补充迁移、隔离集成测试、浏览器模拟验收及角色授权边界修复。
132 行
4.8 KiB
Python
132 行
4.8 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 Permission, Role, RolePermission, UserRole
|
|
from app.schemas.role import RoleQuery
|
|
from app.services.permission_catalog import CATALOG
|
|
|
|
# ==================== 读取 ====================
|
|
|
|
|
|
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 count_permissions_by_role(session: Session, role_ids: list[int]) -> dict[int, int]:
|
|
"""批量统计角色已配置的权限，避免列表逐行查询。"""
|
|
if not role_ids:
|
|
return {}
|
|
stmt = (
|
|
select(RolePermission.role_id, func.count())
|
|
.join(Permission, Permission.id == RolePermission.permission_id)
|
|
.where(RolePermission.role_id.in_(role_ids), Permission.perm_code.in_(CATALOG))
|
|
.group_by(RolePermission.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()
|