409e26614f
- update_receipt 取 _ensure_shipment_of_business 返回值组装出参 （shipment_id 已赋为 payload 新值，与 get_with_refs 重查同源）， 省一次三表 join；update_shipment 同理取锁定行的 business_no - 删 business_repository.count_items（无任何调用的死代码，删除保护 实际走 has_docs） - submit_change 校验段定义与岗位各一条 IN 批量预载（逐条查询在批次 扩大时是写侧 N+1）；岗位字典键 casefold 归一保持与单条版 ci 校对 等价的命中语义；旧值留痕保持逐条（窗口函数优化留待 M3 一并设计） Co-Authored-By: Claude Code <noreply@anthropic.com>
103 lines
3.8 KiB
Python
103 lines
3.8 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_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,
|
||
*,
|
||
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()
|