feat(storage-release): filter pending list by QC work order completion

Add qcWorkOrderCompleted query param and EXISTS-based sample_code filter
so pending reviews can default to QC-complete samples only.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-07-16 12:11:25 +08:00
1 parent 747b8f1c06
commit 1ec3d0d775
3 files changed
+78 -1

No files matched your search

@@ -43,6 +43,7 @@ async def list_storage_release_reviews(
reviewConclusion: Optional[str] = None,
storageStatus: Optional[str] = None,
reviewPhase: Optional[str] = None,
qcWorkOrderCompleted: Optional[bool] = Query(None),
db: Session = Depends(get_db),
user: dict = Depends(require_menu_code("biz:storageReleaseReview:list")),
):
@@ -58,6 +59,7 @@ async def list_storage_release_reviews(
storage_status=storageStatus,
keyword=keyword,
review_phase=reviewPhase,
qc_work_order_completed=qcWorkOrderCompleted,
user_info=user,
)
return success_response(result)
+55 -1
View File
@@ -4,7 +4,7 @@ import math
from datetime import datetime
from typing import Any, Dict, List, Optional
from sqlalchemy import func, or_
from sqlalchemy import and_, exists, func, or_
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
@@ -133,6 +133,42 @@ def normalize_review_phase(review_phase: Optional[str]) -> Optional[str]:
return None
def normalize_qc_work_order_completed(raw) -> Optional[bool]:
"""Return True/False when recognizable; None means no filter."""
if raw is None:
return None
if isinstance(raw, bool):
return raw
s = str(raw).strip().lower()
if s in ("true", "1", "yes"):
return True
if s in ("false", "0", "no"):
return False
return None
def _qc_work_order_exists_for_review_sample(
tenant_no: str, *, unfinished_only: bool = False
):
"""Correlate biz_qc_work_order to BizStorageReleaseReview.sample_code."""
conds = [
BizQcWorkOrder.tenant_no == tenant_no,
BizQcWorkOrder.del_flag == 0,
BizQcWorkOrder.sample_code.isnot(None),
BizQcWorkOrder.sample_code != "",
BizQcWorkOrder.sample_code == BizStorageReleaseReview.sample_code,
BizStorageReleaseReview.sample_code.isnot(None),
BizStorageReleaseReview.sample_code != "",
]
if unfinished_only:
conds.append(BizQcWorkOrder.work_order_status.in_(("dpz", "ypz")))
return (
exists()
.where(and_(*conds))
.correlate(BizStorageReleaseReview)
)
def list_reviews(
db: Session,
*,
@@ -144,6 +180,7 @@ def list_reviews(
storage_status: Optional[str] = None,
keyword: Optional[str] = None,
review_phase: Optional[str] = None,
qc_work_order_completed=None,
user_info=None,
) -> Dict[str, Any]:
q0 = db.query(BizStorageReleaseReview).filter(
@@ -190,6 +227,23 @@ def list_reviews(
if lab_codes:
kw_conds.append(BizStorageReleaseReview.sample_code.in_(lab_codes))
q0 = q0.filter(or_(*kw_conds))
qc_done = normalize_qc_work_order_completed(qc_work_order_completed)
if qc_done is True:
q0 = q0.filter(
_qc_work_order_exists_for_review_sample(tenant_no, unfinished_only=False),
~_qc_work_order_exists_for_review_sample(tenant_no, unfinished_only=True),
)
elif qc_done is False:
q0 = q0.filter(
or_(
~_qc_work_order_exists_for_review_sample(
tenant_no, unfinished_only=False
),
_qc_work_order_exists_for_review_sample(
tenant_no, unfinished_only=True
),
)
)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizStorageReleaseReview, user_info, db, tenant_no)
@@ -0,0 +1,21 @@
from app.crud.storage_release_review import normalize_qc_work_order_completed
def test_normalize_qc_work_order_completed_accepts_bool_and_strings():
assert normalize_qc_work_order_completed(True) is True
assert normalize_qc_work_order_completed(False) is False
assert normalize_qc_work_order_completed("true") is True
assert normalize_qc_work_order_completed("TRUE") is True
assert normalize_qc_work_order_completed("1") is True
assert normalize_qc_work_order_completed("yes") is True
assert normalize_qc_work_order_completed("false") is False
assert normalize_qc_work_order_completed("0") is False
assert normalize_qc_work_order_completed("no") is False
assert normalize_qc_work_order_completed(" FALSE ") is False
def test_normalize_qc_work_order_completed_rejects_unknown():
assert normalize_qc_work_order_completed(None) is None
assert normalize_qc_work_order_completed("") is None
assert normalize_qc_work_order_completed("all") is None
assert normalize_qc_work_order_completed("maybe") is None