FastAPI+SQLAlchemy嵌套事务报错：无法操作已关闭事务

报错原因 #

1. 外层会话生命周期管理错误：get_session中使用with cached_sessionmaker.begin() as session，该上下文会在yield返回会话后，请求结束时自动提交/回滚并关闭会话。但你在接口循环里多次调用session.commit()，会提前提交外层事务，导致会话进入关闭状态，后续嵌套事务操作触发Can't operate on a closed transaction错误。
2. 嵌套事务处理冗余：session.begin_nested()作为上下文管理器，本身会在代码块正常执行时提交子事务，异常时自动回滚子事务，无需手动调用session.rollback()，手动回滚反而会干扰外层事务状态。
3. 变量类型错误：uploaded_items和failed_items被定义为列表，却用+=1计数，会引发TypeError，应改用整数计数器。

解决方法 #

修正会话管理逻辑、嵌套事务用法和变量类型，确保单个条目失败不影响其他条目提交：

router = APIRouter()

@lru_cache()
def get_session_maker() -> sessionmaker:
    engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=True)
    return sessionmaker(engine)

def get_session() -> Generator[Session, None, None]:
    cached_sessionmaker = get_session_maker()
    session = cached_sessionmaker()  # 手动管理会话，不自动开启事务
    try:
        yield session
    finally:
        session.close()

@router.post("/items")
def add_items(
    session: Session = Depends(get_session),
) -> Dict[str, Any]:

    request_inputs = [
        RequestInput(name="chair", used_for="sitting"),
        RequestInput(name="table", used_for="dining"),
        RequestInput(name="tv", used_for="watching"),
    ]

    uploaded_count = 0
    failed_count = 0

    try:
        for request_input in request_inputs:
            try:
                with session.begin_nested():  # 嵌套事务上下文自动处理子事务
                    item = Item(**request_input.dict())
                    session.add(item)
                    session.refresh(item)
                    uploaded_count += 1
            except IntegrityError:
                failed_count += 1
            except Exception:
                failed_count += 1
        
        session.commit()  # 所有成功的子事务统一提交到数据库
    except Exception as e:
        session.rollback()  # 外层异常时回滚所有操作
        raise
    finally:
        session.close()

    return {
        "uploaded": uploaded_count,
        "failed": failed_count,
    }

关键修正说明 #

• 会话管理：get_session不再自动开启事务，将事务控制权交还给接口逻辑，避免会话提前关闭。
• 嵌套事务：利用begin_nested()上下文自动处理子事务的提交/回滚，单个条目失败仅回滚自身子事务，不影响其他条目。
• 事务提交：循环结束后统一提交外层事务，确保所有成功的子事务被持久化。
• 异常处理：外层添加全局异常捕获，确保意外情况时回滚所有操作，避免数据不一致。

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