SQLAlchemy 2.0(Python)异步事务回滚失效：如何实现正确事务控制？

你的问题核心是误用了嵌套事务（begin_nested()），且未正确完成事务的提交/回滚流程，导致异常时无法回滚到初始状态。以下是针对你的场景的正确实现方案：

方案一：用async with自动管理事务（推荐） #

这种方式无需手动处理提交和回滚，上下文管理器会自动完成事务的生命周期：

async def actualize_objects(objects, object_repository) -> None:
    async with object_repository._session.begin():
        to_create, to_update = [], []
        await object_repository.archive_all()
        already_have = await object_repository.get_all_ids()
        for obj in objects:
            if obj.id not in already_have:
                to_create.append(obj)
            else:
                to_update.append(obj)

        await object_repository.create_all(to_create)
        await object_repository.update_all(to_update)

• 代码块正常执行完毕时，上下文管理器会自动提交事务
• 任何异常抛出时，会自动回滚所有数据库操作，包括archive_all()的修改

方案二：手动管理事务流程（适合需要细粒度控制的场景） #

如果需要自定义异常处理逻辑，可以手动控制事务的开启、提交和回滚：

async def actualize_objects(objects, object_repository) -> None:
    session = object_repository._session
    await session.begin()
    try:
        to_create, to_update = [], []
        await object_repository.archive_all()
        already_have = await object_repository.get_all_ids()
        for obj in objects:
            if obj.id not in already_have:
                to_create.append(obj)
            else:
                to_update.append(obj)

        await object_repository.create_all(to_create)
        await object_repository.update_all(to_update)
        await session.commit()
    except Exception as error:
        print('ошибка')
        await session.rollback()
        raise  # 必须重新抛出异常，让上层调用感知错误

关键注意事项 #

• 避免用begin_nested()处理全流程事务：嵌套事务是为局部回滚设计的（基于保存点），不适合你这种需要整体回滚的场景，应该使用顶级事务session.begin()。
• 手动管理时必须调用commit()：之前的代码缺少提交步骤，即使无异常，数据库修改也不会持久化；异常时的rollback()也需要在事务开启的前提下才能生效。
• 异常后重新抛出：捕获异常后若不重新抛出，上层逻辑会误以为操作成功，导致数据不一致。
• 确保所有操作复用同一个session：你的object_repository已持有_session，只要所有数据库方法都基于该session，事务就能覆盖所有步骤。

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