SQLAlchemy异步查询事务回滚问题解析与解决

2024-12-16 07:25:25

SQLAlchemy异步连接查询成功但事务回滚问题解析与解决

当使用SQLAlchemy异步会话执行只读查询时，即便查询成功且会话已关闭，数据库仍回滚相关事务，这无疑会给开发者带来困扰。本文将深入剖析此问题的原因，并提供行之有效的解决方案。

问题根源

SQLAlchemy异步会话默认会在async with块结束后回滚事务，即使没有显式地进行写操作。这是因为异步上下文管理器为了保证资源安全释放，会自动执行事务管理操作。若只读查询放在一个事务块内，即便查询本身没有修改数据，事务结束时也会进行回滚。

另外，日志信息中的INFO sqlalchemy.engine.Engine ROLLBACK也直接表明了数据库引擎执行了回滚操作。即便应用层显示查询成功并关闭了会话，但由于事务管理的机制，数据库层面的回滚动作依然会发生。

解决方案

以下提供几种解决方案，可根据实际情况选择合适的方法。

1. 显式提交事务

对于只读操作，可以在async with块结束前显式提交事务。这可以告诉数据库本次事务已完成，无需回滚。

代码示例:

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

async def execute_read_query(session_maker, query):
    async with session_maker() as session:
        result = await session.execute(query)
        # 提交事务
        await session.commit()
        data = result.fetchall()
        return data

# 异步引擎和会话创建
engine = create_async_engine("数据库连接字符串", echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)

# 查询语句示例
query = select(YourModel)  # 假设 YourModel 是已定义的 SQLAlchemy 模型

# 执行查询
data = asyncio.run(execute_read_query(async_session, query))
print(data)

操作步骤：

确保已安装SQLAlchemy及其异步扩展：pip install sqlalchemy[asyncio]。
使用create_async_engine创建异步引擎。
使用async_sessionmaker创建会话工厂。
在async with块内执行查询。
查询完成后，使用await session.commit()显式提交事务。

2. 使用`begin_nested` 或 `begin_detached`创建子事务

可以将只读操作放在子事务中执行。子事务的提交或回滚不会影响到父事务。

begin_nested代码示例：

async def execute_read_query_nested_transaction(session_maker, query):
    async with session_maker() as session:
        # 开始一个嵌套事务
        async with session.begin_nested():
            result = await session.execute(query)
            data = result.fetchall()
            # 注意：这里不需要显式commit, 因为父事务管理提交/回滚
        return data

# 使用方法类似上面显式提交事务的方式
data = asyncio.run(execute_read_query_nested_transaction(async_session, query))
print(data)

begin_detached 代码示例：

async def execute_read_query_detached_transaction(session_maker, query):
    async with session_maker() as session:
        # 开始一个分离事务
        async with session.begin_detached():
            result = await session.execute(query)
            data = result.fetchall()
            # 注意：这里也不需要显式commit, 分离事务独立管理
        return data

# 使用方法类似上面显式提交事务的方式
data = asyncio.run(execute_read_query_detached_transaction(async_session, query))
print(data)

操作步骤：

与显式提交事务类似，需创建异步引擎和会话工厂。
在async with块内，使用async with session.begin_nested(): 或 async with session.begin_detached(): 创建一个嵌套/分离事务块。
在事务块内执行查询。
嵌套事务的提交和回滚由父事务管理，分离事务则独立管理，因此不需要在子事务块内显式提交或回滚。

3. 设置`autocommit`选项

另一种方法是将数据库连接设置为自动提交模式。在这种模式下，每个 SQL 语句都会被视为一个独立的事务，执行后会自动提交。

代码示例：

# 创建异步引擎时设置autocommit=True
engine = create_async_engine("数据库连接字符串?async_fallback=True", echo=True, execution_options={"autocommit": True})

async_session = async_sessionmaker(engine, expire_on_commit=False)

async def execute_read_query_autocommit(session_maker, query):
    async with session_maker() as session:
        result = await session.execute(query)
        data = result.fetchall()
        return data

# 使用方法类似上面显式提交事务的方式
data = asyncio.run(execute_read_query_autocommit(async_session, query))
print(data)

操作步骤：

创建异步引擎时，通过execution_options={"autocommit": True} 设置自动提交选项。
后续查询操作将自动提交，无需显式管理事务。

安全提示: 开启自动提交模式后，务必确认只读操作不会对数据完整性产生影响，因为它绕过了事务的ACID特性。同时，请确保只对真正的只读查询使用此选项。对于包含数据修改的操作，必须使用显式事务管理。

4. 使用 `autoflush=False` 关闭自动刷新

有时候SQLAlchemy 会在事务提交前自动刷新（flush）会话，将挂起的更改写入数据库，这也可能触发不必要的事务。可以在创建会话时关闭自动刷新。

代码示例：

async_session = async_sessionmaker(engine, expire_on_commit=False, autoflush=False)

async def execute_read_query_autoflush(session_maker, query):
    async with session_maker() as session:
        result = await session.execute(query)
        data = result.fetchall()
    # 会话关闭时自动提交
    return data

data = asyncio.run(execute_read_query_autoflush(async_session, query))
print(data)

操作步骤：

创建异步会话工厂时，设置 autoflush=False。
此时，会话将不会自动刷新，避免不必要的事务行为。当async with块结束后，会话会自动提交事务（因为没有显式启用事务）。

总结

当SQLAlchemy异步连接查询成功但数据库回滚事务时，通常是因为事务管理不当造成的。通过显式提交事务、使用子事务、设置自动提交模式或关闭自动刷新，可以有效解决这个问题。选择哪种方案取决于具体的应用场景和需求。开发者应仔细权衡各种方案的优劣，并根据实际情况选择最合适的方案，以确保数据库操作的正确性和效率。

相关资源：