本文探讨了在SQLAlchemy 2.0中，使用DB-API风格的绑定参数执行原始SQL语句时遇到的常见ArgumentError问题，特别是当参数包含日期时间对象时。文章详细解释了该错误的原因，并提供了解决方案：利用sql_conn.exec_driver_sql()方法，该方法能直接将SQL命令和参数传递给底层DB-API驱动，从而确保参数正确绑定，有效避免了参数类型或结构不匹配的问题，确保了代码的健壮性和安全性。
1. 问题背景与挑战
在sqlalchemy这样的orm框架中，虽然我们通常倾向于使用其高级抽象来构建查询，但在某些特定场景下，例如执行数据库特有的功能、进行性能优化，或者处理复杂的批量操作时，直接执行原始sql语句是不可避免的。此时，为了防止sql注入攻击并正确处理不同数据类型，使用参数绑定（db-api风格）是最佳实践。
然而，在使用SQLAlchemy 2.0版本时，尝试通过 sql_conn.execute(sqlalchemy.sql.text(command), params) 结合DB-API风格的问号占位符 (?) 和一个包含日期时间对象的元组作为参数时，可能会遇到 sqlalchemy.exc.ArgumentError: List argument must consist only of tuples or dictionaries 这样的错误。尽管 params 变量显然是一个元组，但SQLAlchemy的 execute 方法在处理 text() 构造和参数列表时，对参数的结构有特定的预期，尤其是在期望进行批量操作或更复杂的参数映射时。当参数是一个简单的元组，且包含特定类型（如 datetime）时，这种结构上的不匹配就可能导致上述错误。
2. 解决方案：使用 exec_driver_sql()
SQLAlchemy 2.0 提供了一个更直接、更符合DB-API原生行为的方法来解决这个问题：sql_conn.exec_driver_sql()。这个方法允许用户直接将原始SQL字符串和参数列表传递给底层的DB-API驱动连接，由驱动本身负责参数的绑定和语句的执行。这绕过了 sqlalchemy.sql.text() 在 execute 方法中可能进行的额外解析和参数结构验证，从而避免了 ArgumentError。
示例代码
以下是使用 exec_driver_sql() 解决上述问题的示例代码：
from datetime import datetime, timedeltaimport sqlalchemy# 数据库连接字符串，请根据实际情况修改# 示例为SQL Server通过ODBC驱动连接db_con_string = 'Driver={ODBC Driver 17 for SQL Server};Server=tcp:your_server.database.windows.net,1433;Database=your_database;Uid=your_user;Pwd=your_password;'connection_url = sqlalchemy.engine.URL.create("mssql+pyodbc", query={"odbc_connect": db_con_string})engine = sqlalchemy.create_engine(connection_url)# 定义要删除的日期阈值：90天前days_to_keep = 90threshold_date = datetime.utcnow() + timedelta(days=-days_to_keep)# SQL 命令，使用DB-API风格的问号占位符command = 'DELETE myschema.Logs WHERE [DateTimeSent] < ?'# 参数列表，确保是一个元组，即使只有一个参数params = (threshold_date,)try: with engine.begin() as sql_conn: # 使用 exec_driver_sql 执行原始SQL和绑定参数 result = sql_conn.exec_driver_sql(command, params) print(f"成功删除 {result.rowcount} 条记录。")except sqlalchemy.exc.SQLAlchemyError as e: print(f"执行SQL时发生错误: {e}")except Exception as e: print(f"发生未知错误: {e}")# 示例：验证删除操作（可选）# with engine.connect() as conn:# remaining_logs = conn.exec_driver_sql('SELECT COUNT(*) FROM myschema.Logs').scalar()# print(f"剩余日志数量: {remaining_logs}")
代码解析
导入必要的模块: datetime, timedelta 用于日期时间计算；sqlalchemy 用于数据库交互。构建数据库连接: 使用 sqlalchemy.engine.URL.create 和 sqlalchemy.create_engine 建立与SQL Server的连接。请务必替换 db_con_string 中的占位符为您的实际数据库凭据。定义SQL命令和参数:command 变量存储了要执行的原始SQL DELETE 语句。请注意，它使用了DB-API风格的问号 (?) 作为参数占位符。这种占位符是许多DB-API 2.0兼容驱动（如 pyodbc）的标准。params 变量被定义为一个元组 (threshold_date,)。即使只有一个参数，也建议将其包装在元组中，以符合DB-API对参数列表的期望。执行SQL:with engine.begin() as sql_conn: 创建了一个事务上下文。在这个块中执行的所有操作都将作为一个原子单元提交或回滚。sql_conn.exec_driver_sql(command, params) 是核心。它直接将SQL命令字符串和参数元组传递给底层的数据库驱动，由驱动负责参数的正确绑定和语句的执行。result.rowcount 可以获取受影响的行数。
3. exec_driver_sql() 与 execute(text()) 的区别
理解这两种方法的使用场景至关重要：
sql_conn.execute(sqlalchemy.sql.text(command), parameters):
这是SQLAlchemy中执行文本SQL的常用方式。sqlalchemy.sql.text() 将原始SQL字符串封装成一个可被SQLAlchemy理解的文本构造。execute() 方法会通过SQLAlchemy的内部机制来处理参数绑定。它期望参数以特定的格式提供，例如：当SQL使用命名参数 (:param_name) 时，parameters 应为字典（{“param_name”: value}）。当进行批量操作时，parameters 应为字典或元组的列表（[{“col1”: val1}, {“col1”: val2}] 或 [(val1,), (val2,)]）。当参数是一个简单的元组（如 (datetime_object,)）且与 text() 结合时，SQLAlchemy的 execute 方法可能会因为其内部参数处理逻辑而产生 ArgumentError，因为它可能期望一个更复杂的结构或命名参数。
sql_conn.exec_driver_sql(command, parameters):
这个方法是SQLAlchemy 2.0 中引入的，旨在提供一个直接的“通道”来与底层DB-API驱动交互。它不涉及SQLAlchemy对SQL字符串或参数的额外解析或转换。它直接将 command 和 parameters 传递给底层DB-API连接对象的 execute() 方法。因此，command 必须使用底层DB-API驱动所支持的参数占位符（例如，pyodbc 使用 ?，psycopg2 使用 %s，cx_Oracle 使用 :param）。parameters 必须是底层DB-API驱动所期望的参数格式，通常是一个元组或列表。适用于需要最大程度控制原始SQL执行、处理特定驱动行为或绕过SQLAlchemy高级抽象的场景。
4. 注意事项与最佳实践
SQL注入防护: 始终使用参数绑定来传递动态值，切勿直接将用户输入拼接进SQL字符串中。exec_driver_sql() 和 execute(text()) 都支持参数绑定，是安全实践。数据库驱动兼容性: exec_driver_sql() 的参数占位符（如 ?、%s、:param）取决于你使用的底层DB-API驱动。例如，pyodbc 通常使用 ?，psycopg2 (PostgreSQL) 使用 %s，而 cx_Oracle 使用命名参数或位置参数（如 :1）。请查阅你所用驱动的文档。事务管理: 始终使用 with engine.begin() as conn: 或 with engine.connect() as conn: conn.begin() 来管理事务，确保数据的一致性。错误处理: 捕获 sqlalchemy.exc.SQLAlchemyError 或更具体的异常类型，以便优雅地处理数据库操作中可能出现的错误。SQLAlchemy 2.0 风格: exec_driver_sql() 是SQLAlchemy 2.0 推荐的用于直接执行原始SQL的方法之一，与 engine.execute() 或 connection.execute() 的行为有所区别。在SQLAlchemy 2.0中，engine.execute() 和 connection.execute() 更倾向于处理SQLAlchemy表达式或 text() 构造，并期望参数以字典形式传递，尤其是在与 text() 结合时。
总结
在SQLAlchemy 2.0中，当需要执行带有DB-API风格绑定参数的原始SQL语句，并且遇到 ArgumentError 时，特别是当参数包含日期时间等复杂类型时，sql_conn.exec_driver_sql() 方法是解决此问题的首选方案。它提供了一个直接且高效的途径，将SQL命令和参数传递给底层数据库驱动，确保了参数的正确绑定和语句的顺利执行。理解 exec_driver_sql() 与 execute(text()) 的区别，并根据具体需求选择合适的方法，是编写健壮、安全且高效的SQLAlchemy应用的关键。
以上就是在SQLAlchemy中正确使用DB-API风格的绑定参数执行SQL语句的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1368640.html
微信扫一扫 
支付宝扫一扫 
