44 lines
1.3 KiB
Python
44 lines
1.3 KiB
Python
"""三平台外部单协作共用的归档/作废可见性规则。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
ARCHIVE_SCOPES: tuple[str, ...] = ("active", "archived", "all")
|
|
|
|
|
|
def normalize_archive_scope(value: str | None) -> str:
|
|
"""归档可见范围：默认只看活跃行。"""
|
|
scope = str(value or "active").strip().lower()
|
|
return scope if scope in ARCHIVE_SCOPES else "active"
|
|
|
|
|
|
def apply_order_visibility(
|
|
query: Any,
|
|
model: Any,
|
|
*,
|
|
archive_scope: str,
|
|
include_void: bool,
|
|
void_status: str = "void",
|
|
) -> Any:
|
|
"""给协作台账查询统一追加归档和作废过滤。"""
|
|
scope = normalize_archive_scope(archive_scope)
|
|
if scope == "active":
|
|
query = query.filter(model.archived_at.is_(None))
|
|
elif scope == "archived":
|
|
query = query.filter(model.archived_at.isnot(None))
|
|
if not include_void:
|
|
query = query.filter(model.status != void_status)
|
|
return query
|
|
|
|
|
|
def count_archived(db: Session, model: Any) -> int:
|
|
"""统计全部已归档行，不受当前归档视图过滤影响。"""
|
|
return int(
|
|
db.query(func.count(model.id)).filter(model.archived_at.isnot(None)).scalar()
|
|
or 0
|
|
)
|