如何动态创建pypika.Criterion？多表查询过滤实现方案咨询

问题背景 #

我定义了用于查询过滤的数据类Filters如下：

class Filters:
   table: str
   column: str
   value: List[str]
   operation: Operator

当需要从数据库多表加载数据时，会接收包含多表过滤条件的filters: List[Filters]列表。我想要实现如下逻辑，但pypika.Criterion并无add方法：

from pypika import Query, Table, Criterion

def build_query(table: str, columns: List[str], filters: List[Query]):
   criterion = pypika.Criterion.any([])
   for filter in filters:
      if filter.table == table:
          criterion.add(Table(table).field(filter.column) operator filter.values)
   return Query._from(table).select(','.join(columns).where(criterion).get_sql(quote_char=None)

请问有什么可行的解决方案？

可行解决方案 #

Pypika的Criterion是通过逻辑运算符（&、|）来组合条件的，没有add方法，我们可以通过逐步累积条件对象的方式实现需求，同时修正原代码中的语法和类型错误，具体方案如下：

核心思路 #

1. 初始化空条件变量，用None或Criterion.all([])作为初始值
2. 遍历过滤规则，仅处理与当前表匹配的条件
3. 根据operation枚举值，生成对应的Pypika条件对象
4. 用逻辑运算符（&表示与，|表示或）逐步组合条件
5. 最后将组合好的条件传入查询的where方法，无匹配条件则跳过该方法

修正后的完整代码 #

假设Operator是包含常见查询操作的枚举类型，代码示例：

from typing import List, Enum
from pypika import Query, Table

class Operator(Enum):
    EQ = "eq"
    IN = "in"
    GT = "gt"
    LIKE = "like"

class Filters:
    table: str
    column: str
    value: List[str]
    operation: Operator

def build_query(table: str, columns: List[str], filters: List[Filters]):
    target_table = Table(table)
    criterion = None
    
    for filter_item in filters:
        if filter_item.table != table:
            continue
        
        col = target_table.field(filter_item.column)
        # 根据操作符生成对应查询条件
        match filter_item.operation:
            case Operator.EQ:
                cond = col == filter_item.value[0]
            case Operator.IN:
                cond = col.in_(filter_item.value)
            case Operator.GT:
                cond = col > filter_item.value[0]
            case Operator.LIKE:
                cond = col.like(filter_item.value[0])
            case _:
                continue
        
        # 累积条件：首次赋值，后续用逻辑符组合
        if criterion is None:
            criterion = cond
        else:
            criterion &= cond  # 逻辑与，需逻辑或则替换为 `|`
    
    # 构建最终查询
    query = Query.from_(target_table).select(*columns)
    if criterion is not None:
        query = query.where(criterion)
    
    return query.get_sql(quote_char=None)

关键说明 #

• 条件组合逻辑：示例用&实现多条件“与”查询，若需要“或”逻辑，直接替换为|即可
• 操作符扩展：可根据业务需求添加更多Operator枚举值，对应Pypika字段支持的方法（如<=、>=、not_like等）
• 空条件处理：当没有匹配当前表的过滤规则时，不会生成WHERE子句，符合常规查询逻辑

内容的提问来源于stack exchange，提问作者antek16x