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
|