PyMySQL执行多条查询时rollback回滚失效，建表操作未撤销如何解决？

• 核心原因是MySQL中所有DDL语句（包括CREATE TABLE、ALTER TABLE、DROP TABLE等库表结构操作语句）执行后会触发隐式提交，不受手动事务控制。你执行第一条CREATE TABLE语句时，当前事务就已经被系统自动提交，后续即便触发异常执行rollback，也无法回滚DDL执行前的状态。
• 你疑惑的“未调用commit但操作已提交”也是同样原因：DDL的隐式提交不需要依赖代码中的connection.commit()调用，执行后直接将变更持久化到数据库。
• 调整后添加的connection.begin()仅能显式开启普通事务，无法绕过DDL的隐式提交规则，所以修改后依然无法回滚表创建操作。

方案1：更换测试语句验证事务逻辑 #

如果只是要测试事务回滚能力，更换为支持事务的DML操作（INSERT、UPDATE、DELETE）即可，示例代码如下：

import pymysql

def function():
    # 显式指定autocommit=False，避免环境配置差异影响
    connection = pymysql.connect(
        host='你的数据库地址',
        user='用户名',
        password='密码',
        database='数据库名',
        autocommit=False
    )
    try:
        cursor = connection.cursor()
        # 测试用的test表提前在库中创建好，不要放在事务内执行DDL
        q1 = "INSERT INTO test_table (column1) VALUES ('测试数据')"
        cursor.execute(q1)
        # 故意执行错误语句触发异常
        q2 = "INSERT INTO 不存在的表 (column1) VALUES ('测试数据2')"
        cursor.execute(q2)
        
        connection.commit()
    except pymysql.Error as e:
        print("触发异常")
        print(e)
        try:
            connection.rollback()
            print("回滚成功，第一条INSERT操作已撤销")
        except:
            print("回滚失败")
            pass
    finally:
        connection.close()
                   
function()

方案2：添加DDL补偿逻辑 #

如果业务场景确实需要在流程中执行DDL，MySQL原生不支持DDL回滚，需要自行实现补偿逻辑：在异常触发时手动回滚之前的DDL操作，示例如下：

import pymysql

def function():
    connection = pymysql.connect(
        host='你的数据库地址',
        user='用户名',
        password='密码',
        database='数据库名',
        autocommit=False
    )
    table_created_flag = False
    try:
        connection.begin()
        cursor = connection.cursor()
        q1 = "CREATE TABLE 0_test_table (column1 varchar(255))"
        cursor.execute(q1)
        table_created_flag = True # 标记表已创建
        
        q2 = "CREATE TABLE 0_test_table (column1 varchar(255))"
        cursor.execute(q2)
        
        connection.commit()
    except pymysql.Error as e:
        print("触发异常")
        print(e)
        try:
            connection.rollback()
            # 补偿逻辑：手动删除已创建的表
            if table_created_flag:
                cursor.execute("DROP TABLE IF EXISTS 0_test_table")
                print("补偿逻辑执行完成：测试表已删除")
        except Exception as err:
            print("回滚或补偿逻辑执行失败")
            print(err)
            pass
    finally:
        cursor.close()
        connection.close()
               
function()

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