Python3+SQLAlchemy构建防注入查询遇参数格式错误求助

错误原因 #

你传入的param是普通列表类型，而SQLAlchemy的session.execute()方法要求序列类型的参数必须是元组，或者字典/字典列表，普通列表不符合参数格式要求，因此抛出该错误。

解决方案 #

方案1：将参数列表转为元组

直接修改param的类型为元组即可：

data = [
    {"department_id": "vegies", "country": "Mexico"},
    {"department_id": "vegies", "country": "Australia"},
    {"department_id": "beef", "country": "Australia"}
]

sql = """SELECT * FROM products WHERE (department_id = %s AND country = %s) OR (department_id = %s AND country = %s) OR (department_id = %s AND country = %s)"""

param = ('vegies', 'Mexico', 'vegies', 'Australia','beef', 'Australia')

res = session.execute(text(sql), param).fetchall()

方案2：动态生成查询（适配任意数量条件）

手动拼接多个OR条件效率低且易出错，推荐利用SQL的(列1, 列2) IN (...)语法动态生成查询，同时保持参数化查询避免注入：

from sqlalchemy import text

data = [
    {"department_id": "vegies", "country": "Mexico"},
    {"department_id": "vegies", "country": "Australia"},
    {"department_id": "beef", "country": "Australia"}
]

# 将字典列表转换为元组的元组
condition_tuples = tuple((item["department_id"], item["country"]) for item in data)
# 使用命名参数传递条件，语法更清晰
sql = text("SELECT * FROM products WHERE (department_id, country) IN :conditions")

res = session.execute(sql, {"conditions": condition_tuples}).fetchall()

这种方式无需手动编写多个OR分支，无论data中有多少条条件都能自动适配，同时完全规避SQL注入风险。

方案3：使用SQLAlchemy ORM构造查询（推荐）

如果项目使用SQLAlchemy ORM（假设已定义Product模型对应products表），可以直接用ORM的条件构造器生成查询，彻底避免手写SQL：

from sqlalchemy import or_, and_
# 导入你的Product模型
from models import Product

data = [
    {"department_id": "vegies", "country": "Mexico"},
    {"department_id": "vegies", "country": "Australia"},
    {"department_id": "beef", "country": "Australia"}
]

# 构造每个(department_id, country)的条件组
condition_groups = []
for item in data:
    group = and_(
        Product.department_id == item["department_id"],
        Product.country == item["country"]
    )
    condition_groups.append(group)

# 用OR连接所有条件组并执行查询
query = session.query(Product).filter(or_(*condition_groups))
results = query.all()

这种方式由SQLAlchemy自动生成安全的参数化SQL，不仅避免注入，还能获得ORM带来的类型安全和代码可维护性。

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