57 lines
2.2 KiB
Python
57 lines
2.2 KiB
Python
"""归属变更留痕的共享数据访问（客户/业务两维度共用）。
|
||
|
||
owner_type 与引用列由调用方参数化 —— 此前客户、业务两侧各持一套逐行
|
||
同构的 list/count/add，排序口径与 join 兜底的调整须同步两处；收口后
|
||
「被删人员留痕仍可读」「翻页稳定双键倒序」等口径单点维护。
|
||
"""
|
||
|
||
from sqlalchemy import Row, func, select
|
||
from sqlalchemy.orm import Session, aliased
|
||
|
||
from app.models import OwnershipChangeLog, Personnel
|
||
|
||
|
||
def list_by_ref(
|
||
session: Session,
|
||
*,
|
||
owner_type: int,
|
||
ref_column,
|
||
ref_id: int,
|
||
offset: int,
|
||
limit: int,
|
||
) -> list[Row]:
|
||
"""查某引用对象的归属留痕，行内容为 (留痕, 旧归属人名, 新归属人名)。
|
||
|
||
人员删除后 join 出 None，留痕固化 ID 兜底可读。按 (create_time, id)
|
||
双键倒序：create_time 是秒级精度，同一秒内多条排序不确定，id 兜底
|
||
保证翻页稳定（与操作日志同一理由）。
|
||
"""
|
||
old_person = aliased(Personnel)
|
||
new_person = aliased(Personnel)
|
||
stmt = (
|
||
select(OwnershipChangeLog, old_person.personnel_name, new_person.personnel_name)
|
||
.outerjoin(old_person, OwnershipChangeLog.old_owner_personnel_id == old_person.id)
|
||
.outerjoin(new_person, OwnershipChangeLog.new_owner_personnel_id == new_person.id)
|
||
.where(OwnershipChangeLog.owner_type == owner_type, ref_column == ref_id)
|
||
.order_by(OwnershipChangeLog.create_time.desc(), OwnershipChangeLog.id.desc())
|
||
.offset(offset)
|
||
.limit(limit)
|
||
)
|
||
return list(session.execute(stmt).all())
|
||
|
||
|
||
def count_by_ref(session: Session, *, owner_type: int, ref_column, ref_id: int) -> int:
|
||
"""统计某引用对象的归属留痕总数，供分页使用。"""
|
||
stmt = select(func.count(OwnershipChangeLog.id)).where(
|
||
OwnershipChangeLog.owner_type == owner_type,
|
||
ref_column == ref_id,
|
||
)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def add(session: Session, log: OwnershipChangeLog) -> OwnershipChangeLog:
|
||
"""写入一条归属变更留痕。只 flush 不 commit，与主表归属更新同事务提交。"""
|
||
session.add(log)
|
||
session.flush()
|
||
return log
|