如何在SQLAlchemy中实现多SQL语句执行失败时批量回滚？

方案1：自动事务上下文（推荐） #

SQLAlchemy 1.4及以上版本提供engine.begin()上下文管理器，会自动处理事务生命周期：进入上下文时开启事务，所有语句执行成功则自动提交，任意语句失败则回滚所有已执行操作。

修改你的示例代码如下：

from sqlalchemy import create_engine, text

engine = create_engine("sqlite:///foo.db")

s1 = text("CREATE TABLE mytable (x int, y int)")
s2 = text("ALTER TABLE myothertable ADD z int;")

# 用engine.begin()替代engine.connect()
with engine.begin() as conn:
    conn.execute(s1)
    conn.execute(s2)

当s2因myothertable不存在执行失败时，s1创建表的操作会被自动回滚，数据库不会留下mytable。

方案2：手动管理事务 #

如果需要更精细的逻辑控制（比如加日志、条件判断），可以手动开启、提交或回滚事务：

from sqlalchemy import create_engine, text

engine = create_engine("sqlite:///foo.db")

s1 = text("CREATE TABLE mytable (x int, y int)")
s2 = text("ALTER TABLE myothertable ADD z int;")

with engine.connect() as conn:
    trans = conn.begin()
    try:
        conn.execute(s1)
        conn.execute(s2)
        trans.commit()  # 所有语句成功才提交
    except Exception:
        trans.rollback()  # 任意失败则回滚
        raise  # 可选：重新抛出异常以便上层处理

关于多语句放同一个text的问题 #

将多条SQL语句塞进单个text()执行出现不同步错误，是因为部分数据库驱动（如SQLite默认驱动）不支持单execute()调用执行多语句，或是SQLAlchemy的text()对多语句处理存在限制。

最优解决方式仍是拆分语句+事务管理，而非强行合并语句。若确实需要执行多语句，可使用exec_driver_sql()（绕过部分SQLAlchemy抽象，需确保语句符合数据库语法），但依然要配合事务保证原子性：

from sqlalchemy import create_engine

engine = create_engine("sqlite:///foo.db")

multi_stmt = """
CREATE TABLE mytable (x int, y int);
ALTER TABLE mytable ADD z int;
"""

with engine.begin() as conn:
    conn.exec_driver_sql(multi_stmt)

不过这种方式兼容性不如拆分语句方案，优先推荐拆分语句+事务的组合。

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