diff --git a/backend/app/services/bx/base.py b/backend/app/services/bx/base.py index 489dad637..e4f267be2 100644 --- a/backend/app/services/bx/base.py +++ b/backend/app/services/bx/base.py @@ -45,6 +45,7 @@ class BxCrudService: *, user: Optional["BxUserContext"] = None, skip_scope: bool = False, + order_by: Optional[Any] = None, ) -> Tuple[Sequence[M], int]: query = db.query(model).filter(model.tenant_no == tenant_no) if hasattr(model, "del_flag"): @@ -54,7 +55,9 @@ class BxCrudService: if user and not skip_scope: query = apply_bx_scope(query, model, user, db, tenant_no) total = query.count() - if hasattr(model, "sort_order"): + if order_by is not None: + query = query.order_by(order_by) + elif hasattr(model, "sort_order"): query = query.order_by(model.sort_order.asc(), model.create_time.desc()) else: query = query.order_by(model.create_time.desc()) diff --git a/backend/app/services/bx/list_sort.py b/backend/app/services/bx/list_sort.py new file mode 100644 index 000000000..e5343fd53 --- /dev/null +++ b/backend/app/services/bx/list_sort.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from typing import Optional + +from app.models.bx_models import ( + BxAdvanceApply, + BxInvoice, + BxInvoiceLedger, + BxReimburseApply, +) +from app.services.crm.list_sort import resolve_datetime_order_by + +ADVANCE_WHITELIST = { + "applyTime": BxAdvanceApply.create_time, + "issueTime": BxAdvanceApply.issue_time, +} +REIMBURSE_WHITELIST = { + "applyTime": BxReimburseApply.create_time, + "payTime": BxReimburseApply.pay_time, +} +INVOICE_WHITELIST = {"createTime": BxInvoice.create_time} +LEDGER_WHITELIST = {"createTime": BxInvoiceLedger.create_time} + + +def order_advance(sort_field: Optional[str], sort_order: Optional[str], default=None): + if default is None: + default = BxAdvanceApply.create_time.desc() + return resolve_datetime_order_by(sort_field, sort_order, ADVANCE_WHITELIST, default) + + +def order_reimburse(sort_field: Optional[str], sort_order: Optional[str], default=None): + if default is None: + default = BxReimburseApply.create_time.desc() + return resolve_datetime_order_by(sort_field, sort_order, REIMBURSE_WHITELIST, default) + + +def order_invoice(sort_field: Optional[str], sort_order: Optional[str], default=None): + if default is None: + default = BxInvoice.create_time.desc() + return resolve_datetime_order_by(sort_field, sort_order, INVOICE_WHITELIST, default) + + +def order_ledger(sort_field: Optional[str], sort_order: Optional[str], default=None): + if default is None: + default = BxInvoiceLedger.create_time.desc() + return resolve_datetime_order_by(sort_field, sort_order, LEDGER_WHITELIST, default) + + +def parse_sort_params(params: dict) -> tuple: + """从 request/body dict 取 sortField/sortOrder(兼容 snake)。""" + sf = params.get("sortField") or params.get("sort_field") + so = params.get("sortOrder") or params.get("sort_order") + return (str(sf).strip() if sf else None), (str(so).strip() if so else None) diff --git a/backend/tests/test_bx_list_sort.py b/backend/tests/test_bx_list_sort.py new file mode 100644 index 000000000..1314369bd --- /dev/null +++ b/backend/tests/test_bx_list_sort.py @@ -0,0 +1,52 @@ +# backend/tests/test_bx_list_sort.py +from app.models.bx_models import BxAdvanceApply, BxInvoice, BxInvoiceLedger, BxReimburseApply +from app.services.bx.list_sort import ( + order_advance, + order_invoice, + order_ledger, + order_reimburse, + parse_sort_params, +) + + +def test_advance_apply_time_asc(): + expr = order_advance("applyTime", "asc") + assert "create_time" in str(expr).lower() + assert "asc" in str(expr).lower() + + +def test_advance_issue_time(): + expr = order_advance("issueTime", "desc") + assert "issue_time" in str(expr).lower() + + +def test_advance_invalid_falls_back(): + default = BxAdvanceApply.create_time.desc() + assert order_advance("hack", "asc", default=default) is default + + +def test_reimburse_pay_time(): + expr = order_reimburse("payTime", "desc") + assert "pay_time" in str(expr).lower() + + +def test_invoice_create_time(): + expr = order_invoice("createTime", "asc") + assert "create_time" in str(expr).lower() + + +def test_ledger_create_time(): + expr = order_ledger("createTime", "desc") + assert "create_time" in str(expr).lower() + + +def test_parse_sort_params_camel(): + sf, so = parse_sort_params({"sortField": "applyTime", "sortOrder": "asc"}) + assert sf == "applyTime" + assert so == "asc" + + +def test_parse_sort_params_snake(): + sf, so = parse_sort_params({"sort_field": "payTime", "sort_order": "desc"}) + assert sf == "payTime" + assert so == "desc"