liuwenxinandClaude Code
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>
111 lines
4.2 KiB
Python
111 lines
4.2 KiB
Python
"""业务岗位角色数据访问。
|
|
|
|
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
|
"""
|
|
|
|
from sqlalchemy import Select, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import BusinessRole
|
|
from app.schemas.business_role import BusinessRoleQuery
|
|
|
|
# ==================== 读取 ====================
|
|
|
|
|
|
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_by_codes(session: Session, role_codes: list[str]) -> dict[str, BusinessRole]:
|
|
"""按编码批量取岗位（参数批次校验段用，消除逐条 N+1）。
|
|
|
|
返回 {编码: 岗位}。MySQL ci 校对下编码大小写变形也能命中，
|
|
键取库里的规范 role_code，调用方据此把维度值归一为规范值。
|
|
"""
|
|
if not role_codes:
|
|
return {}
|
|
rows = session.execute(
|
|
select(BusinessRole).where(BusinessRole.role_code.in_(role_codes))
|
|
).scalars()
|
|
return {role.role_code: role for role in rows}
|
|
|
|
|
|
def list_roles(session: Session, query: BusinessRoleQuery) -> list[BusinessRole]:
|
|
"""分页查询岗位（含停用）。排序 (sort_order, id)，与系统角色口径一致。"""
|
|
stmt = _apply_filters(select(BusinessRole), query)
|
|
stmt = (
|
|
stmt.order_by(BusinessRole.sort_order, BusinessRole.id)
|
|
.offset(query.offset)
|
|
.limit(query.page_size)
|
|
)
|
|
return list(session.execute(stmt).scalars())
|
|
|
|
|
|
def count_roles(session: Session, query: BusinessRoleQuery) -> int:
|
|
"""统计符合条件的岗位总数，供分页使用。"""
|
|
stmt = _apply_filters(select(func.count(BusinessRole.id)), query)
|
|
return int(session.execute(stmt).scalar_one())
|
|
|
|
|
|
def list_all(session: Session) -> list[BusinessRole]:
|
|
"""全部岗位（含停用，不分页）——岗位下拉选项的取数口径。
|
|
|
|
岗位规模有限（内置 + 自定义），一次取全量，不受分页上限约束；
|
|
排序与 list_roles 一致（sort_order, id），下拉与列表相对次序相同。
|
|
"""
|
|
stmt = select(BusinessRole).order_by(BusinessRole.sort_order, BusinessRole.id)
|
|
return list(session.execute(stmt).scalars())
|
|
|
|
|
|
def _apply_filters(stmt: Select, query: BusinessRoleQuery) -> Select:
|
|
"""应用通用筛选条件。列表与计数共用，避免两处口径不一致。
|
|
|
|
关键词走 ``contains(autoescape=True)``（转义 LIKE 通配符）——岗位编码含下划线，
|
|
不转义时 ``_`` 会当单字符通配执行。
|
|
"""
|
|
if query.keyword:
|
|
stmt = stmt.where(
|
|
or_(
|
|
BusinessRole.role_code.contains(query.keyword, autoescape=True),
|
|
BusinessRole.role_name.contains(query.keyword, autoescape=True),
|
|
)
|
|
)
|
|
if query.status is not None:
|
|
stmt = stmt.where(BusinessRole.status == query.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()
|