本文介绍如何使用 python 的 `sql` 包，根据 json 中定义的条件字典列表（含字段名、操作符和值），安全、动态地构造 sql 查询的 where 子句，避免拼接原始 sql，兼顾可读性与防注入能力。
在使用 sql 包构建结构化查询时，手动为每个条件编写 Column == value 表达式虽可行，但难以扩展且易出错。理想方案是将条件逻辑抽象为可复用、可配置的处理流程。核心思路是：将字符串操作符（如 "="、"!="）映射为对应的 Python 比较函数，并对每个条件字典生成一个布尔表达式对象，最后用逻辑与（&）串联所有条件。
为此，我们借助标准库 operator 模块提供的一组函数（如 operator.eq、operator.ne 等），并配合 functools.reduce 实现多条件的链式组合。以下是 get_cond 方法的完整实现：
import operator
from functools import reduce
from typing import List, Dict, Any
def get_cond(self, filters: List[Dict[str, Any]]) -> Any:
# 支持常见 SQL 比较操作符到 Python 函数的映射
op_map = {
"=": operator.eq,
"!=": operator.ne,
"<": operator.lt,
">": operator.gt,
"<=": operator.le,
">=": operator.ge,
"IN": lambda col, val: col.isin(val) if isinstance(val, (list, tuple)) else col == val,
"LIKE": lambda col, val: col.like(val),
}
# 逐条解析 filters，生成 sql.Column 比较表达式
conditions = []
for cond_dict in filters:
col_name = cond_dict.get("COND_COL")
op_str = cond_dict.get("COND_OP", "=").strip()
val = cond_dict.get("COND_VAL")
if not col_name or op_str not in op_map:
raise ValueError(f"Invalid condition: {cond_dict}")
column = sql.Column(self, col_name)
try:
# 调用映射函数生成比较表达式（如 operator.eq(column, "VAL1")）
expr = op_map[op_str](column, val)
conditions.append(expr)
except Exception as e:
raise ValueError(f"Failed to build condition {cond_dict}: {e}")
# 使用 & 连接所有条件；空列表返回 True（即无 where 约束）
if not conditions:
return sql.SQL(True) # 或直接返回 None，取决于 sql 包对 where=None 的处理
return reduce(operator.and_, conditions)✅ 关键说明：sql.Column(self, col_name) 确保列引用绑定到当前表对象，保持上下文安全；op_map 可按需扩展（如增加 BETWEEN、IS NULL 等），推荐封装为类属性或配置常量；IN 和 LIKE 等特殊操作符需单独处理，因其语法不同于二元比较（例如 isin() 是 sql 包原生方法）；使用 reduce(operator.and_, conditions) 等价于 conditions[0] & conditions[1] & ...，语义清晰且可扩展性强；异常处理保障输入数据异常时快速失败，便于调试与日志追踪。
最后，在主逻辑中只需一行调用即可完成集成：
# 替换原有硬编码条件 stmt.where = obj.get_cond(input["CONDS"])
该方案完全规避了字符串格式化或 f-string 拼接 SQL，所有参数均通过 sql 包的表达式对象传递，底层会自动处理占位符（如 %s）与参数绑定，天然具备 SQL 注入防护能力。同时，结构清晰、易于单元测试与维护，是构建动态查询服务的理想实践。
立即学习“Python免费学习笔记（深入）”；
