如何让FastAPI路径操作函数实现事务自动提交与回滚

核心思路是修改数据库会话的依赖生成器get_db，利用FastAPI依赖注入的特性，在路径操作函数执行完成后自动处理事务的提交或回滚：

修改数据库会话依赖函数 #

把原来的get_db改成以下形式，通过try-except捕获路径函数执行时的异常，自动完成事务的提交/回滚：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from typing import Generator

engine = create_engine(settings.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
        # 路径操作函数正常执行完毕，提交事务
        db.commit()
    except Exception:
        # 捕获任何异常，回滚事务
        db.rollback()
        # 重新抛出异常，让FastAPI正常处理HTTP响应
        raise
    finally:
        # 无论事务成功还是失败，最终关闭会话
        db.close()

路径操作函数的简化 #

修改后，所有路径操作函数无需手动调用db.commit()，只需要正常操作数据库即可。以你的用户更新接口为例：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

@app.put("/users/{id}")
def update_user(
    id: int,
    new_name: str,
    new_login_id: str,
    db: Session = Depends(get_db),
):
    user = db.query(User).filter(User.id == id).first()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    user.name = new_name
    
    if db.query(User).filter(User.login_id == new_login_id).first():
        # 抛出异常后，事务会自动回滚，name的修改不会生效
        raise HTTPException(status_code=409, detail="Already used login id")

    user.login_id = new_login_id
    # 无需手动commit，依赖函数会自动处理
    return {"message": "User updated successfully"}

为什么这个方案可行？ #

FastAPI的依赖注入机制中，yield之前的代码会在路径函数执行前运行，yield之后的代码会在路径函数执行完成（无论成功还是抛出异常）后运行：

• 如果路径函数正常结束，没有抛出异常，yield后的db.commit()会执行，提交所有数据库操作
• 如果路径函数抛出任何异常（包括HTTPException），except块会触发，执行db.rollback()回滚事务，然后重新抛出异常让FastAPI返回对应的HTTP错误响应
• finally块保证会话最终会被关闭，避免资源泄漏

为什么之前的尝试不生效？ #

• 中间件无法直接获取当前请求绑定的数据库会话，因为会话是通过依赖注入到路径函数的，中间件的执行时机早于依赖注入
• 在finally块中添加db.commit()会忽略异常，无论路径函数是否出错都会提交事务，不符合原子性要求
• 全局错误处理器无法获取当前请求的数据库会话，每个请求的会话都是独立生成的，处理器无法关联到对应会话

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