使用SQLAlchemy session.execute执行多语句原生查询的异常处理

核心原因 #

Python的MySQL驱动（包括SQLAlchemy封装的驱动）在执行多语句时，默认只会处理第一个语句的结果，后续语句的错误会被隐藏，直到显式遍历所有执行结果或主动查询会话错误。

方法1：遍历执行结果（SQLAlchemy场景） #

如果使用SQLAlchemy的session.execute执行多语句，返回的CursorResult（Result对象）是惰性迭代的，只需遍历所有结果就能触发后续语句的异常：

result = session.execute("UPDATE users SET email = test@test.com; UPDATE users SET wrong_column = test@test.com;")
# 遍历所有结果，触发潜在的异常
for _ in result:
    pass

当执行到错误语句时，遍历过程会直接抛出对应的数据库异常。

方法2：主动查询会话错误 #

无论使用哪种驱动，都可以在多语句执行后，通过MySQL内置的SHOW ERRORS命令查询当前会话的所有错误：

# 执行多语句
session.execute("UPDATE users SET email = test@test.com; UPDATE users SET wrong_column = test@test.com;")

# 查询错误
error_cursor = session.execute("SHOW ERRORS")
errors = error_cursor.fetchall()

if errors:
    # 提取错误信息并抛出异常
    error_msg = "; ".join([f"[{err[0]}] {err[2]}" for err in errors])
    raise Exception(f"多语句执行失败: {error_msg}")

注意：SHOW ERRORS会返回当前会话中所有未被清除的错误，需确保在目标多语句执行后立即调用，避免其他操作的错误干扰。

方法3：驱动原生多语句处理（原生驱动场景） #

如果使用原生驱动（如mysql-connector-python、PyMySQL），启用multi=True参数执行多语句，然后遍历结果迭代器：

import mysql.connector

conn = mysql.connector.connect(user='xxx', password='xxx', database='xxx')
cursor = conn.cursor()

try:
    # 启用multi=True执行多语句
    results = cursor.execute("UPDATE users SET email = test@test.com; UPDATE users SET wrong_column = test@test.com;", multi=True)
    # 遍历每个语句的结果，触发异常
    for res in results:
        # 对于更新类语句，可以检查影响行数，确认执行状态
        if res.with_rows:
            res.fetchall()
finally:
    cursor.close()
    conn.close()

这些方法都不需要拆分SQL语句，也无需安装额外库，完美适配含RETURN ... ;、END;的函数定义场景。

内容的提问来源于stack exchange，提问作者Basiti