certflow.views.bases.datasources.base_data 源代码

"""基础数据数据源（客户端过滤 + 排序）

统一的全量加载 → 客户端过滤 → 客户端排序 → 客户端分页实现，供材质牌号 /
口径映射 / 型号→压力 三张基础表复用，避免每表各写一遍。
"""

from __future__ import annotations

from typing import Any

from certflow.views.bases.query_data_source import (
    BaseQueryDataSource,
    client_filter_records,
    client_sort_records,
    export_records_to_csv,
)




class BaseDataDataSource(BaseQueryDataSource):
    """基础数据（全量小表）数据源：list_all 后客户端过滤/排序/分页。"""

    supports_server_sort = False
    supports_server_paging = False
    quick_search_or = False
    editable = False
    correction = False

    def __init__(
        self,
        controller: Any,
        columns: list[tuple[str, str]],
        filter_columns: list[dict] | None = None,
        label: str = "基础数据",
    ) -> None:
        self._controller = controller
        self._columns = list(columns)
        self._filter_columns = filter_columns or [{"label": h, "field": f} for h, f in columns]
        self.label = label



    def get_columns(self) -> list[tuple[str, str]]:
        return list(self._columns)




    def get_filter_columns(self) -> list[dict]:
        return list(self._filter_columns)




    def query(
        self,
        conditions: dict,
        page: int,
        page_size: int,
        order_by: str | None = None,
        order_desc: bool = False,
        or_filters: Any = None,
    ) -> dict[str, Any]:
        records = self._controller.list_all()
        records = client_filter_records(records, conditions)
        records = client_sort_records(records, order_by, order_desc)
        return self.client_paginate(records, page, page_size, conditions)




    def export_csv(self, path: str) -> int:
        return export_records_to_csv(self._controller.list_all(), self.get_columns(), path)