SQLAlchemy 2.0.3 抛出 (ArgumentError)List 参数必须仅包含元组或字典

Question

我之前使用 SQLAlchemy 1.4.44 和 pandas 1.5.1，并且执行 SQL 存储过程的以下代码有效：

sql_connection: str = "driver={ODBC Driver 17 for SQL Server};server=localhost\SQLEXPRESS;database=Finance;trusted_connection=yes"
sql_engine: sqlalchemy.engine = sqlalchemy.create_engine(
    url=sqlalchemy.engine.URL.create(drivername="mssql+pyodbc", query={"odbc_connect": sql_connection})
)
with sql_engine.connect() as connection:
    query: str = "EXEC dbo.get_latest_tickers @etf=?"
    return pandas.read_sql_query(sql=query, con=connection, params=[etf])

我切换到 SQLAlchemy 2.0.3 和 pandas 1.5.3，并通过使用 sqlalchemy.text 包装对存储过程的调用来修改代码，因为此版本的 SQLAlchemy 需要它。修改后的代码如下：

sql_connection: str = "driver={ODBC Driver 17 for SQL Server};server=localhost\SQLEXPRESS;database=Finance;trusted_connection=yes"
sql_engine: sqlalchemy.engine = sqlalchemy.create_engine(
    url=sqlalchemy.engine.URL.create(drivername="mssql+pyodbc", query={"odbc_connect": sql_connection})
)
with sql_engine.connect() as connection:
    query: str = sqlalchemy.text("EXEC dbo.get_latest_tickers @etf=?")
    return pandas.read_sql_query(sql=query, con=connection, params=[etf])

代码抛出以下异常：

(ArgumentError)List argument must consist only of tuples or dictionaries

我尝试按如下方式修改 params 参数，但修改也失败：

return pandas.read_sql_query(sql=query, con=connection, params={"@etf": etf})

抛出的异常如下：

{DBAPIError}(pyodbc.Error)('07002', '[07002] [Microsoft][ODBC Driver 17 for SQL Server)COUNT field incorrect or syntax error (0) (SQLExecDirectW)')

如何传递参数来执行存储过程？

Answer 1

我了解到函数 sqlalchemy.text 为绑定参数提供了后端中立的支持。此类参数必须采用命名列格式。请参阅https://www.tutorialspoint.com/sqlalchemy/sqlalchemy_core_using_textual_sql.htm。

以下修改后的代码有效：

with sql_engine.connect() as connection:
    query: str = sqlalchemy.text("EXEC dbo.get_latest_tickers @etf=:etf")
    return pandas.read_sql_query(sql=query, con=connection, params={"etf": etf})