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

Question

I was previously using SQLAlchemy 1.4.44 with pandas 1.5.1 and the following code that executes a SQL Stored procedure worked:

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])

I switched to SQLAlchemy 2.0.3 and pandas 1.5.3 and revised the code by wrapping the call to stored procedure with sqlalchemy.text as this version of SQLAlchemy requires it. The revised code is as follows:

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])

The code throws the following exception:

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

I have tried revising params argument as follows but the revision also fails:

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

The exception thrown is as follows:

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

How do I pass parameters to execute the stored procedure?

Answer 1

I learned that the function sqlalchemy.text provides backend-neutral support for bind parameters. Such parameters must be in the named column format. See https://www.tutorialspoint.com/sqlalchemy/sqlalchemy_core_using_textual_sql.htm.

The following revised code works:

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})