如何将列表绑定到 SQLAlchemy 中自定义查询中的参数？答案

【问题标题】：How can I bind a list to a parameter in a custom query in SQLAlchemy?如何将列表绑定到 SQLAlchemy 中自定义查询中的参数？
【发布时间】：2021-12-05 20:56:51
【问题描述】：

出于性能原因，我使用此 SQL：

 sql_tmpl = """delete from Data where id_data in (:iddata) """
 params = {
                    'iddata':[1, 2,3 4],
                    }

 # 'session' is a session object from SQLAlchemy
 self.session.execute(text(sql_tmpl), params)

但是，我遇到了一个异常：

NotSupportedError: (NotSupportedError) ('Python type list not supported.  param=1', 'HY097')

是否有一种解决方法可以让我将列表绑定到“in”子句的参数？

【问题讨论】：

传递给 IN 的值在 SQL 中仍然是离散的、呈现的值，因此 SQL 语句本身仍然必须形成包含单独的绑定参数。如果您使用的是文字 SQL，则需要对 ":iddata1, :iddata2, .." 等进行字符串连接。下面 Gary 的答案是正确的。

标签： python sqlalchemy

【解决方案1】：

一种适用于任何数据库的新方法（不仅仅是依赖 psycopg2 的类型适应）使用扩展绑定参数：

sql_tmpl = """delete from Data where id_data in :iddata"""
params = { 'iddata': [1, 2, 3, 4], }
# session is a session object from sqlalchemy
t = text(sql_tmpl)
t = t.bindparams(bindparam('iddata', expanding=True))
self.session.execute(t, params)

【讨论】：

使用 pymysql 和 SQLAlchemy 1.2，这不是必需的。 expanding 功能似乎与语句缓存有关。更多细节在这里：github.com/sqlalchemy/sqlalchemy/issues/3953
@jnylen 那是因为pymysql支持列表适配。扩展 bindparams 不仅仅适用于烘焙（缓存）查询，因为并非所有 DB-API 驱动程序都支持列表自适应，在这种情况下，您必须手动将所需的占位符格式化并格式化为查询。在某些查询中它也明显更快，例如将一个巨大的列表传递给in_()：stackoverflow.com/questions/57829682/…

【解决方案2】：

psycopg2 现在支持type adaptation，它允许将列表传递到查询中的单个参数化值中。这也适用于 SQLAlchemy，至少对于 PostgreSQL 数据库的原始 SQL 式查询（我无权访问其他数据库类型，所以我不知道 sqlalchemy 是否会尊重其他数据库的此约定，但我的倾向^需要引用是它会起作用）。

some_ids = [1, 2, 3, 4]
query = "SELECT * FROM my_table t WHERE t.id = ANY(:ids);"
conn.execute(sqlalchemy.text(query), ids=some_ids)
## runs just fine

我发现如果没有对 sqlalchemy.text 的包装器调用，它会给出 ProgrammingError: syntax error at or near ":"。

【讨论】：

【解决方案3】：

试试不带括号的:iddata。这一直对我有用。

sql_tmpl = """delete from Data where id_data in :iddata """

【讨论】：

工作正常，但参数必须是元组。不适用于列表。
取决于使用的 DB-API 驱动程序。

【解决方案4】：

使用元组而不是列表，并且查询中的参数不需要括号：

sql_tmpl = "delete from Data where id_data in :iddata"
params = {  
   'iddata':(1, 2, 3, 4),
}
self.session.execute(text(sql_tmpl), params)

【讨论】：

【解决方案5】：

据我所知，没有一个 SQL 引擎允许传入数组参数。 SQLAlchemy 处理这个问题的方式是为数组中的每个项目传入一个参数。

>>> from sqlalchemy.sql import table, column
>>> print(table('Data').delete(column('id_data').in_([5, 6, 7,])))
DELETE FROM "Data" WHERE id_data IN (:id_data_1, :id_data_2, :id_data_3)

如果您不使用 SQLAlchemy 表达式构造，则需要手动执行此操作。

【讨论】：

不知道写答案的时候是不是这样，但现在psycopg2支持bunch of coercions, including arrays。参见例如Tuples adaptation。不幸的是，SQL Alchemy 似乎使用了自己的逻辑，因此该功能仅在直接使用 psycopg2 时可用。

【解决方案6】：

添加到dwanderson's answer：使用SQLAlchemy，您可以使用func 方法将“any”函数添加到查询中。这适用于我使用 SQLAlchemy 1.0.9 和 PostgreSQL 数据库。

通用示例：

from sqlalchemy import func

# some list
id_list = [1, 2, 3]

# assuming you have created a session
query = session.query(Table.user_name, Table.user_id).\
    filter(Table.user_id == func.any(id_list))

# one way of running the query
query.all()

您可以验证列表是作为单个参数传递的（而不是列表中每个对象的参数）。

print(query.statement)

选择用户 ID、用户名从表 WHERE table.user_id = any(:any_1)

【讨论】：

【解决方案7】：

您可以使用循环生成 where 子句并使用 ** 在 query.execute 参数中拆分列表。这是一个例子：https://gist.github.com/pawl/555e5eecce77d4de0ada

【讨论】：

在 SQL 中进行程序化？缺点是什么？

【解决方案8】：

在 Microsoft SQL Server 中，您可以使用表值参数来完成同样的事情。

SELECT * FROM table_name WHERE customer_id in (SELECT * FROM @MyTVP)

TVP 目前仅支持PyTDS，不支持 PyODBC。如果您有较新版本的 SQLAlchemy，expanding=True 标志 detailed by Jason Damiani 可能是最好的方法。但是 TVP 会在紧要关头完成。

【讨论】：

【解决方案9】：

如果您正在处理原始 SQL，您可以采用这种方法：

ids = [1000032, 1000048]
sql = 'SELECT CompNo, CompName, CompType FROM Component WHERE DeptID IN %(ids)s' % {"ids": tuple(ids)}
cursor.execute(sql)

【讨论】：