certflow.views.bases.query_data_source module

查询数据源协议与通用引擎（阶段 A 抽象层）

定义 QueryDataSource 协议，使 BaseQueryView 能以统一方式驱动不同数据源 （销售计划 / 校正队列 / 基础数据 / BOM 查询），无需在视图里写 if 分支。

能力标志（capability flags）决定 UI 行为： - supports_server_sort：列头点击走「重新查询」（销售计划 / 校正队列）。 - supports_server_paging：分页由服务端承担（销售计划 / 校正队列）。 - quick_search_or：是否支持跨字段 OR 快查（仅销售计划）。 - editable / correction：是否暴露行内编辑 / 校正动作（客户端源按需在子类开启）。

客户端数据源（基础数据 / BOM）复用本模块的 client_filter_records / client_sort_records 在内存中过滤排序，避免给每个表写一遍逻辑。

certflow.views.bases.query_data_source.client_filter_records(records, conditions)[源代码]

客户端条件过滤。

conditions 形如 {field: {"value": v, "operator": op}}，支持的 operator： contains / eq / ne / gte / lte / gt / lt / between``(value=[lo,hi]) / ``is_null / is_not_null。

参数:
返回类型:

list[Any]

certflow.views.bases.query_data_source.client_sort_records(records, order_by, order_desc=False)[源代码]

客户端排序：None 恒排末尾；同列类型同质，数字按数值、其余按字符串。

参数:
返回类型:

list[Any]

certflow.views.bases.query_data_source.export_records_to_csv(records, columns, path)[源代码]

把记录按列定义导出为 CSV（UTF-8-BOM，Excel 友好）。返回行数。

参数:
返回类型:

int

class certflow.views.bases.query_data_source.QueryDataSource(*args, **kwargs)[源代码]

基类：Protocol

查询数据源协议（结构化契约，供 BaseQueryView 驱动）。

supports_server_sort: bool
supports_server_paging: bool
quick_search_or: bool
editable: bool
correction: bool
query(conditions, page, page_size, order_by=None, order_desc=False, or_filters=None)[源代码]

返回 {records, page, page_size, total, total_pages, conditions}。

参数:

  • conditions (dict)

  • page (int)

  • page_size (int)

  • order_by (str | None)

  • order_desc (bool)

  • or_filters (Any)

返回类型:

dict[str, Any]

get_columns()[源代码]

返回 [(表头, 字段), ...]。

返回类型:

list[tuple[str, str]]

export_csv(path)[源代码]

导出当前结果到 CSV，返回行数。

参数:

path (str)

返回类型:

int

class certflow.views.bases.query_data_source.BaseQueryDataSource[源代码]

基类：ABC

数据源基类：提供能力标志默认值与默认筛选列推导。

supports_server_sort: bool = False
supports_server_paging: bool = False
quick_search_or: bool = False
editable: bool = False
correction: bool = False
abstractmethod query(conditions, page, page_size, order_by=None, order_desc=False, or_filters=None)[源代码]
参数:

  • conditions (dict)

  • page (int)

  • page_size (int)

  • order_by (str | None)

  • order_desc (bool)

  • or_filters (Any)

返回类型:

dict[str, Any]

abstractmethod get_columns()[源代码]
返回类型:

list[tuple[str, str]]

abstractmethod export_csv(path)[源代码]
参数:

path (str)

返回类型:

int

get_filter_columns()[源代码]

默认把所有显示列作为可筛选字段；子类可覆盖以精简。

返回类型:

list[dict]

client_paginate(records, page, page_size, conditions)[源代码]

客户端分页：对已过滤/排序的内存 records 切片，返回统一契约。

供 BaseDataDataSource / BomQueryDataSource 等客户端数据源复用， 避免每个子类重复书写 total/total_pages/page_records 切片逻辑。

参数:
返回类型:

dict[str, Any]