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>
97 lines
3.9 KiB
Python
97 lines
3.9 KiB
Python
"""回款数据访问：回款行与关联引用（业务号/发货单号/发货日）。
|
||
|
||
只读写数据并 flush；commit/rollback 与业务异常一律在 service 层。
|
||
|
||
- 列表与详情统一 join 出 ``business_no / shipment_no / ship_date``，
|
||
ship_date 供 service 按 ``formula.payment_days`` 派生回款天数 —— 关联
|
||
引用经本层查询一次供给，service 不再对发货单二次查询
|
||
- shipment_id 是手动指定（可空），发货单关联一律 outerjoin
|
||
"""
|
||
|
||
from datetime import date
|
||
|
||
from sqlalchemy import Select, func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Business, Receipt, Shipment
|
||
from app.schemas.receipt import ReceiptQuery
|
||
|
||
|
||
def get_by_id(session: Session, receipt_id: int) -> Receipt | None:
|
||
"""按主键取回款行（供摘要预取与状态判定；不锁行，写保护锚在业务行）。"""
|
||
return session.get(Receipt, receipt_id)
|
||
|
||
|
||
def get_with_refs(
|
||
session: Session, receipt_id: int
|
||
) -> tuple[Receipt, str, str | None, date | None] | None:
|
||
"""回款行 + 业务号 + 发货单号 + 发货日（DTO 与 payment_days 派生共用）。
|
||
|
||
business 主表为 inner join（FK NOT NULL，业务与回款同存亡）；发货单
|
||
可未指定，outerjoin 后 shipment_no / ship_date 可能为 None。
|
||
"""
|
||
row = session.execute(
|
||
select(Receipt, Business.business_no, Shipment.shipment_no, Shipment.ship_date)
|
||
.join(Business, Business.id == Receipt.business_id)
|
||
.outerjoin(Shipment, Shipment.id == Receipt.shipment_id)
|
||
.where(Receipt.id == receipt_id)
|
||
).first()
|
||
return row
|
||
|
||
|
||
def get_shipment(session: Session, shipment_id: int) -> Shipment | None:
|
||
"""按主键取发货单（同业务强校验的只读预查；FK RESTRICT 为最终防线）。"""
|
||
return session.get(Shipment, shipment_id)
|
||
|
||
|
||
def _apply_filters(stmt: Select, query: ReceiptQuery) -> Select:
|
||
"""回款列表的筛选条件（全部落在 Receipt 自身列上）。"""
|
||
if query.business_id is not None:
|
||
stmt = stmt.where(Receipt.business_id == query.business_id)
|
||
if query.date_from is not None:
|
||
stmt = stmt.where(Receipt.receipt_date >= query.date_from)
|
||
if query.date_to is not None:
|
||
stmt = stmt.where(Receipt.receipt_date <= query.date_to)
|
||
if query.is_barter is not None:
|
||
stmt = stmt.where(Receipt.is_barter == query.is_barter)
|
||
# settlement_status 是 M3 核算防重锚点（需规 2.3.3），必须保持具名字段
|
||
if query.settlement_status is not None:
|
||
stmt = stmt.where(Receipt.settlement_status == query.settlement_status)
|
||
return stmt
|
||
|
||
|
||
def list_receipts(
|
||
session: Session, query: ReceiptQuery
|
||
) -> list[tuple[Receipt, str, str | None, date | None]]:
|
||
"""分页查询回款（附业务号/发货单号/发货日），按 id 倒序。"""
|
||
stmt = _apply_filters(
|
||
select(Receipt, Business.business_no, Shipment.shipment_no, Shipment.ship_date)
|
||
.join(Business, Business.id == Receipt.business_id)
|
||
.outerjoin(Shipment, Shipment.id == Receipt.shipment_id),
|
||
query,
|
||
)
|
||
return list(
|
||
session.execute(
|
||
stmt.order_by(Receipt.id.desc()).offset(query.offset).limit(query.page_size)
|
||
).all()
|
||
)
|
||
|
||
|
||
def count_receipts(session: Session, query: ReceiptQuery) -> int:
|
||
"""回款总数（筛选条件与列表共用同一 apply_filters，防两侧口径漂移）。"""
|
||
stmt = _apply_filters(select(func.count()).select_from(Receipt), query)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def create(session: Session, receipt: Receipt) -> Receipt:
|
||
"""新增回款行。只 flush 以取得自增 id，编号唯一键冲突由 service 兜底。"""
|
||
session.add(receipt)
|
||
session.flush()
|
||
return receipt
|
||
|
||
|
||
def delete(session: Session, receipt: Receipt) -> None:
|
||
"""物理删除回款行（已核算与否的准入判定在 service 层）。"""
|
||
session.delete(receipt)
|
||
session.flush()
|