7a0f3f4b98
- 四层实现：仓储（FOR UPDATE 锁行 + 引用计数钩子预留）/ schemas / service / api 共 7 个端点 - 保护规则：内置岗位禁删禁停、编码创建后不可改、模板须在注册表内（未注册 400） - 写操作全走审计上下文；冒烟验证 7 项（含成功/失败双路径落日志） - 端点：列表/新增/详情/编辑/删除/启停 + meta/templates 模板下拉 Co-Authored-By: Claude Code <noreply@anthropic.com>
89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""业务岗位角色数据访问。
|
||
|
||
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
||
"""
|
||
|
||
from sqlalchemy import Select, func, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import BusinessRole
|
||
|
||
# ==================== 读取 ====================
|
||
|
||
|
||
def get_by_id(session: Session, role_id: int, *, for_update: bool = False) -> BusinessRole | None:
|
||
"""按主键查询。删除校验走 ``for_update`` 锁行（同 role_repository 的理由）。"""
|
||
if for_update:
|
||
stmt = select(BusinessRole).where(BusinessRole.id == role_id).with_for_update()
|
||
return session.execute(stmt).scalar_one_or_none()
|
||
return session.get(BusinessRole, role_id)
|
||
|
||
|
||
def get_by_code(session: Session, role_code: str) -> BusinessRole | None:
|
||
"""按编码查询岗位。编码是参数维度值（dimension_value）的锚点。"""
|
||
stmt = select(BusinessRole).where(BusinessRole.role_code == role_code)
|
||
return session.execute(stmt).scalar_one_or_none()
|
||
|
||
|
||
def list_roles(
|
||
session: Session,
|
||
*,
|
||
offset: int,
|
||
limit: int,
|
||
keyword: str | None = None,
|
||
status: int | None = None,
|
||
) -> list[BusinessRole]:
|
||
"""分页查询岗位（含停用）。排序 (sort_order, id)，与系统角色口径一致。"""
|
||
stmt = _apply_filters(select(BusinessRole), keyword=keyword, status=status)
|
||
stmt = stmt.order_by(BusinessRole.sort_order, BusinessRole.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(BusinessRole.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_(
|
||
BusinessRole.role_code.contains(keyword, autoescape=True),
|
||
BusinessRole.role_name.contains(keyword, autoescape=True),
|
||
)
|
||
)
|
||
if status is not None:
|
||
stmt = stmt.where(BusinessRole.status == status)
|
||
return stmt
|
||
|
||
|
||
def count_personnel_by_role(session: Session, role_ids: list[int]) -> dict[int, int]:
|
||
"""批量统计各岗位下的人员数（删除保护的依据）。
|
||
|
||
② 基础数据管理落地 personnel 表后在此实现真实计数（一次 GROUP BY，
|
||
禁止列表循环内逐行查询）；当前无人员表，恒返回空，删除校验对空字典放行。
|
||
"""
|
||
return {}
|
||
|
||
|
||
# ==================== 写入 ====================
|
||
|
||
|
||
def create(session: Session, role: BusinessRole) -> BusinessRole:
|
||
"""新增岗位。只 flush 以取得自增 id。"""
|
||
session.add(role)
|
||
session.flush()
|
||
return role
|
||
|
||
|
||
def delete(session: Session, role: BusinessRole) -> None:
|
||
"""删除岗位。调用方须先完成保护规则校验。"""
|
||
session.delete(role)
|
||
session.flush()
|