如何用Python sql包基于字典列表动态生成SQL WHERE条件

以下是基于Python sql 包的完整实现，核心是完成get_cond方法，实现从字典列表动态生成WHERE子句条件：

import sql
from typing import List, Dict
from functools import reduce
import operator

class QueryGetter(sql.FromItem):

    def __init__(self, table_name: str):
        self.table = table_name

    def get_columns(self, input_columns: List):
        return [sql.Column(self, col) for col in input_columns]

    # 实现动态生成条件的方法
    def get_cond(self, filters: List[Dict]):
        # 映射SQL操作符到sql.Column对应的魔术方法
        op_map = {
            "=": operator.eq,
            "<": operator.lt,
            ">": operator.gt,
            "!=": operator.ne,
            "<=": operator.le,
            ">=": operator.ge,
            # 可按需扩展其他操作符，如LIKE、IN等
        }

        if not filters:
            # 无过滤条件时返回恒真条件，也可根据需求返回None（不生成WHERE子句）
            return sql.Constant(True)

        # 遍历生成单个条件表达式
        conditions = []
        for filt in filters:
            col = sql.Column(self, filt["COND_COL"])
            op = op_map[filt["COND_OP"]]
            cond = op(col, filt["COND_VAL"])
            conditions.append(cond)

        # 用逻辑AND合并所有条件
        return reduce(operator.and_, conditions)
    
    @property
    def params(self):
        return ()

    def __str__(self):
        return self.table


if __name__ == "__main__":
    input = {
        "TABLE_NAME" : "TABLE",
        "COLS": [
            "COL1",
            "COL2",
        ],
        "CONDS":[
            {
                "COND_COL": "COL3",
                "COND_OP": "=",
                "COND_VAL": "VAL1"
            },
            {
                "COND_COL": "COL4",
                "COND_OP": "<",
                "COND_VAL": "VAL2"
            },
            {
                "COND_COL": "COL5",
                "COND_OP": "!=",
                "COND_VAL": "VAL3"
            },
            {
                "COND_COL": "COL6",
                "COND_OP": ">",
                "COND_VAL": "VAL4"
            }
        ]
    }
    obj = QueryGetter(input["TABLE_NAME"])
    cols = obj.get_columns(input["COLS"])
    
    stmt = obj.select(*cols)
    
    # 替换硬编码逻辑，调用get_cond生成动态条件
    cond = obj.get_cond(input["CONDS"])
    stmt.where = cond
    
    # 执行语句（示例）
    # rows = cursor.execute(stmt, stmt.params).fetchall()

关键逻辑说明 #

1. 操作符映射：通过op_map字典将JSON中的字符串操作符（如=、<）映射到Python operator 模块的函数，这些函数会自动调用sql.Column对应的魔术方法（比如operator.eq对应__eq__，等价于col == val）。
2. 单条件生成：遍历每个过滤条件字典，创建对应的sql.Column对象，再通过映射的操作符函数生成单个条件表达式。
3. 条件合并：使用functools.reduce和operator.and_将所有单条件用逻辑AND（&）连接，完全等价于你硬编码的cond1 & cond2 & ...逻辑。
4. 空条件处理：当没有过滤条件时，返回sql.Constant(True)作为恒真条件，确保查询正常执行；也可根据需求返回None，此时不会生成WHERE子句。

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