SQLAlchemy AsyncSession事务未发BEGIN，ProxySQL下可重复读失效排查

我使用SQLAlchemy的AsyncSession上下文管理器开启事务，执行一次简单查询后，通过其他客户端修改读取到的值，再在事务内重新读取该值。我预期两次读取结果一致——毕竟事务在REPEATABLE READ隔离级别下应该遵循这个规则。

需要说明的是，我是通过ProxySQL连接数据库的：

• 直接连接MySQL时，事务行为符合预期：第一次读取值X，其他客户端修改为Y后，事务内第二次读取仍为X；
• 通过ProxySQL时，查看MySQL日志发现SQLAlchemy从未发送BEGIN;语句，仅设置SET AUTOCOMMIT = 0;，最后发送COMMIT;，导致两次读取结果不一致。

我想知道：有没有办法让SQLAlchemy在事务启动时自动发送BEGIN;？或是ProxySQL与SQLAlchemy的配置存在冲突？

引擎配置 #

engine = create_async_engine(
    f"mysql+aiomysql://{user}:{password}@{host}:{port}{db_string}?charset=utf8mb4",
    echo=False,
    pool_pre_ping=True,
    isolation_level="REPEATABLE READ",
    poolclass=AsyncAdaptedQueuePool,
)

session = async_scoped_session(
    async_sessionmaker(engine, expire_on_commit=False),
    scopefunc=asyncio.current_task,
)

代码（未触发预期事务行为） #

async with session.begin() as transaction:
    # 此时`user`.`money`的值为10
    cursor = await transaction.session.execute(text("SELECT `user`.`money` FROM `user` WHERE `user`.`id` = 398"))
    res = cursor.fetchall()
    
    print("RES:", res)
    # RES: [(10,)]

    # 这15秒休眠期间，通过其他客户端将`user`.`money`改为20
    await asyncio.sleep(15)

    cursor = await transaction.session.execute(text("SELECT `user`.`money` FROM `user` WHERE `user`.`id` = 398"))
    res = cursor.fetchall()
    
    print("RES2:", res)
    # RES2: [(20,)]

代码（显式执行BEGIN后符合预期） #

async with session.begin() as transaction:
    await transaction.session.execute(text("BEGIN"))

    # 此时`user`.`money`的值为10
    cursor = await transaction.session.execute(text("SELECT `user`.`money` FROM `user` WHERE `user`.`id` = 398"))
    res = cursor.fetchall()
    
    print("RES:", res)
    # RES: [(10,)]

    # 这15秒休眠期间，通过其他客户端将`user`.`money`改为20
    await asyncio.sleep(15)

    cursor = await transaction.session.execute(text("SELECT `user`.`money` FROM `user` WHERE `user`.`id` = 398"))
    res = cursor.fetchall()
    
    print("RES2:", res)
    # RES2: [(10,)]

1. 配置SQLAlchemy自动发送BEGIN语句 #

SQLAlchemy针对MySQL默认使用隐式事务模式（依赖SET AUTOCOMMIT=0开启事务），但可以通过以下方式强制开启显式事务：

方式一：在数据库URL中添加参数

修改create_async_engine的连接URL，加入explicit_begin=true参数：

engine = create_async_engine(
    f"mysql+aiomysql://{user}:{password}@{host}:{port}{db_string}?charset=utf8mb4&explicit_begin=true",
    echo=False,
    pool_pre_ping=True,
    isolation_level="REPEATABLE READ",
    poolclass=AsyncAdaptedQueuePool,
)

方式二：通过引擎的execution_options设置

在创建引擎时，通过execution_options指定explicit_begin=True：

engine = create_async_engine(
    f"mysql+aiomysql://{user}:{password}@{host}:{port}{db_string}?charset=utf8mb4",
    echo=False,
    pool_pre_ping=True,
    isolation_level="REPEATABLE READ",
    poolclass=AsyncAdaptedQueuePool,
    execution_options={"explicit_begin": True}
)

该配置会让SQLAlchemy在事务启动时主动发送BEGIN语句，替代原有的隐式事务逻辑，确保事务在REPEATABLE READ隔离级别下的一致性。

2. ProxySQL配置排查 #

ProxySQL的默认配置可能会干扰事务处理，建议检查以下项：

• 查看mysql-connect_server_query_rules，确保没有拦截或改写BEGIN语句的规则；
• 确认mysql-autocommit_false_is_transaction参数设置为true，让ProxySQL正确识别SET AUTOCOMMIT=0开启的事务；
• 如果开启了事务池（transaction pooling），可能导致事务上下文混乱，建议关闭或调整相关配置。

3. 验证事务状态 #

可以在代码中添加逻辑验证事务是否正确开启：

async with session.begin() as transaction:
    # 检查当前事务状态
    status_cursor = await transaction.session.execute(text("SELECT @@tx_isolation, @@autocommit"))
    print("事务状态:", status_cursor.fetchall())
    # 后续业务逻辑...

也可以通过SHOW ENGINE INNODB STATUS查看InnoDB的事务详情，确认隔离级别和事务上下文是否正常。

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