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>
122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
"""发货数据访问：发货单行、明细行与业务号关联引用。
|
||
|
||
只读写数据并 flush；commit/rollback 与业务异常一律在 service 层。
|
||
|
||
- 列表与详情统一 join 出 ``business_no``（业务主表 inner join，FK NOT
|
||
NULL，业务与发货单同存亡）
|
||
- 明细行汇总（sum_item_amount）是发货单 total_amount 重算的唯一数据源，
|
||
业务级汇总仍收口 business_service._recalc_totals
|
||
"""
|
||
|
||
from decimal import Decimal
|
||
|
||
from sqlalchemy import Select, delete, func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Business, Receipt, Shipment, ShipmentItem
|
||
from app.schemas.shipment import ShipmentQuery
|
||
|
||
|
||
def get_by_id(session: Session, shipment_id: int) -> Shipment | None:
|
||
"""按主键取发货单行（供摘要预取与编辑赋值；不锁行，写保护锚在业务行）。"""
|
||
return session.get(Shipment, shipment_id)
|
||
|
||
|
||
def get_with_business_no(session: Session, shipment_id: int) -> tuple[Shipment, str] | None:
|
||
"""发货单行 + 业务号（DTO 与审计摘要共用）。"""
|
||
return session.execute(
|
||
select(Shipment, Business.business_no)
|
||
.join(Business, Business.id == Shipment.business_id)
|
||
.where(Shipment.id == shipment_id)
|
||
).first()
|
||
|
||
|
||
def _apply_filters(stmt: Select, query: ShipmentQuery) -> Select:
|
||
"""发货列表的筛选条件（business_id 空=全量，一期按业务维度查看为主）。"""
|
||
if query.business_id is not None:
|
||
stmt = stmt.where(Shipment.business_id == query.business_id)
|
||
if query.ship_date_from is not None:
|
||
stmt = stmt.where(Shipment.ship_date >= query.ship_date_from)
|
||
if query.ship_date_to is not None:
|
||
stmt = stmt.where(Shipment.ship_date <= query.ship_date_to)
|
||
return stmt
|
||
|
||
|
||
def list_shipments(session: Session, query: ShipmentQuery) -> list[tuple[Shipment, str]]:
|
||
"""分页查询发货单（附业务号），按 id 倒序。"""
|
||
stmt = _apply_filters(
|
||
select(Shipment, Business.business_no).join(Business, Business.id == Shipment.business_id),
|
||
query,
|
||
)
|
||
return list(
|
||
session.execute(
|
||
stmt.order_by(Shipment.id.desc()).offset(query.offset).limit(query.page_size)
|
||
).all()
|
||
)
|
||
|
||
|
||
def count_shipments(session: Session, query: ShipmentQuery) -> int:
|
||
"""发货单总数（筛选条件与列表共用同一 apply_filters，防两侧口径漂移）。"""
|
||
stmt = _apply_filters(
|
||
select(func.count())
|
||
.select_from(Shipment)
|
||
.join(Business, Business.id == Shipment.business_id),
|
||
query,
|
||
)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def list_items(session: Session, shipment_id: int) -> list[ShipmentItem]:
|
||
"""发货明细行（详情回显与总金额重算共用）。"""
|
||
return list(
|
||
session.execute(
|
||
select(ShipmentItem)
|
||
.where(ShipmentItem.shipment_id == shipment_id)
|
||
.order_by(ShipmentItem.id)
|
||
).scalars()
|
||
)
|
||
|
||
|
||
def sum_item_amount(session: Session, shipment_id: int) -> Decimal | None:
|
||
"""明细行金额合计（发货单 total_amount 重算的唯一数据源）。"""
|
||
return session.execute(
|
||
select(func.sum(ShipmentItem.amount)).where(ShipmentItem.shipment_id == shipment_id)
|
||
).scalar_one()
|
||
|
||
|
||
def count_receipt_refs(session: Session, shipment_id: int) -> int:
|
||
"""被回款批次指定的引用数（删除保护依据，receipts.shipment_id）。"""
|
||
return int(
|
||
session.execute(
|
||
select(func.count()).select_from(Receipt).where(Receipt.shipment_id == shipment_id)
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
def create(session: Session, shipment: Shipment) -> Shipment:
|
||
"""新增发货单行。只 flush 以取得自增 id，单号唯一键冲突由 service 兜底。"""
|
||
session.add(shipment)
|
||
session.flush()
|
||
return shipment
|
||
|
||
|
||
def replace_items(session: Session, shipment_id: int, rows: list[ShipmentItem]) -> None:
|
||
"""全量替换明细：先删、立即 flush、再插入（business_repository 同款防御写法）。
|
||
|
||
同一 flush 中 SQLAlchemy 先处理 INSERT 后处理 DELETE，先删后插不落库
|
||
会在有唯一键的表上撞键；shipment_items 虽无唯一键，统一按此写法防御。
|
||
"""
|
||
session.execute(delete(ShipmentItem).where(ShipmentItem.shipment_id == shipment_id))
|
||
session.flush()
|
||
for row in rows:
|
||
row.shipment_id = shipment_id
|
||
session.add(row)
|
||
session.flush()
|
||
|
||
|
||
def delete_with_items(session: Session, shipment_id: int) -> None:
|
||
"""删除发货单及明细行（按依赖序：明细 → 主表，FK RESTRICT）。"""
|
||
session.execute(delete(ShipmentItem).where(ShipmentItem.shipment_id == shipment_id))
|
||
session.execute(delete(Shipment).where(Shipment.id == shipment_id))
|
||
session.flush()
|