我使用Python 3和sqlalchemy动态创建一个查询,选择所有满足条件的产品并避免sql注入问题。
我收到以下错误。 列表参数必须仅包含元组或字典
代码是这样的
data =[{"deparment_id": "vegies", "country": "Mexico"},
{"deparment_id": "vegies", "country": "Australia"},
{"deparment_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()
您已经在使用 SQLAlchemy,所以让它为您构建查询。
import sqlalchemy as sa
engine = sa.create_engine("sqlite:///database.sqlite")
products = sa.Table("products", sa.MetaData(), autoload_with=engine)
data = [{"department_id": "vegies", "country": "Mexico"},
{"department_id": "vegies", "country": "Australia"},
{"department_id": "beef", "country": "Australia"}]
where_cases = []
for case in data:
criteria = [products.c[k] == v for k, v in case.items()]
where_cases.append(sa.and_(*criteria))
where_clause = sa.or_(*where_cases)
qry = sa.select(products).where(where_clause)
engine.echo = True
with engine.begin() as conn:
conn.execute(qry)
"""
SELECT products.department_id, products.country
FROM products
WHERE products.department_id = ? AND products.country = ? OR products.department_id = ? AND products.country = ? OR products.department_id = ? AND products.country = ?
[generated in 0.00050s] ('vegies', 'Mexico', 'vegies', 'Australia', 'beef', 'Australia')
"""