feat(system): 业务岗位角色管理接口与保护规则

- 四层实现:仓储(FOR UPDATE 锁行 + 引用计数钩子预留)/ schemas / service / api 共 7 个端点
- 保护规则:内置岗位禁删禁停、编码创建后不可改、模板须在注册表内(未注册 400)
- 写操作全走审计上下文;冒烟验证 7 项(含成功/失败双路径落日志)
- 端点:列表/新增/详情/编辑/删除/启停 + meta/templates 模板下拉

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxinandClaude Code committed 2026-10-12 16:39:08 +08:00
1 parent 1b3509b20f
commit 7a0f3f4b98
6 files changed
+590

No files matched your search

@@ -0,0 +1,88 @@
"""业务岗位角色数据访问。
本层只 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()