如何在FastAPI RealWorld示例应用中实现事务提交与回滚逻辑？

该项目默认使用SQLAlchemy作为ORM层，原生支持事务能力，我们可以基于数据库会话的生命周期控制事务边界，无需大幅改动现有代码，具体实现逻辑如下：

核心前提 #

项目现有架构通过依赖注入为所有repository层方法传递同一个数据库会话实例，所有数据库操作都复用该会话，这是实现全局事务控制的基础，不需要修改repository层的现有逻辑。

基础实现方式 #

在服务层（或路由层，更推荐放在服务层）用try-except块包裹所有数据库操作，统一控制事务提交和回滚：

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from app.modules.xxx.repositories import UserRepository, ArticleRepository

async def multi_table_business_logic(db: AsyncSession):
    # 所有repository复用同一个数据库会话
    user_repo = UserRepository(db)
    article_repo = ArticleRepository(db)
    try:
        # 按业务顺序执行多表操作
        await user_repo.update_user_info(user_id=1, data={"points": 100})
        await article_repo.create_article(data={"author_id": 1, "title": "新文章"})
        await user_repo.delete_expired_records(user_id=1)
        
        # 所有操作无异常，统一提交事务
        await db.commit()
        return {"code": 0, "msg": "操作成功"}
    except SQLAlchemyError as e:
        # 任意操作触发异常，统一回滚所有已执行的数据库改动
        await db.rollback()
        # 可自定义抛出业务异常，由全局异常处理器统一返回给前端
        raise RuntimeError(f"操作失败，已回滚：{str(e)}")

更优雅的封装方案 #

如果项目中多处需要事务控制，可以封装异步上下文管理器避免重复写try-except块：

from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession

@asynccontextmanager
async def transaction_scope(db: AsyncSession):
    try:
        yield
        await db.commit()
    except Exception as e:
        await db.rollback()
        raise e

使用时直接包裹业务逻辑即可，无需手动处理提交/回滚：

async def multi_table_business_logic(db: AsyncSession):
    user_repo = UserRepository(db)
    article_repo = ArticleRepository(db)
    async with transaction_scope(db):
        await user_repo.update_user_info(user_id=1, data={"points": 100})
        await article_repo.create_article(data={"author_id": 1, "title": "新文章"})
        await user_repo.delete_expired_records(user_id=1)
    return {"code": 0, "msg": "操作成功"}

注意事项 #

• 禁止在repository层单独执行commit/rollback操作，所有事务控制权统一收归上层服务层，避免事务提前提交导致回滚失效。
• 如果项目使用的是同步版本的SQLAlchemy，逻辑完全一致，仅需去掉异步相关的await、async关键字即可。

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