FastAPI中间件中SQLAlchemy事务：commit()未写入数据库

问题背景 #

尝试通过FastAPI中间件创建单例数据库会话以避免路由层依赖注入，本地单进程（Docker+Uvicorn）运行正常，但部署到Linux服务器多Gunicorn Worker环境时，退出上下文管理器的隐式commit()无法将变更写入数据库，无报错但数据不持久化。

用户实现代码：

MiddlewareSession = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@app.middleware("http")
async def create_db_singleton(request: Request, call_next):
    with MiddlewareSession() as session:
        with session.begin():
            SessionLocal(session) # Singleton to be used
            response = await call_next(request)
    session.close() # <- added to attempt force close connections
    return response

问题根源 #

• Gunicorn多进程隔离特性：每个Gunicorn Worker是独立的Python进程，全局单例SessionLocal在不同进程中是完全独立的实例，中间件里绑定的会话无法跨进程共享，路由层实际使用的可能是未关联当前请求会话的实例。
• 会话上下文错误嵌套：with MiddlewareSession() as session已自动管理会话生命周期（含提交、关闭），嵌套with session.begin()会开启嵌套事务，但路由中使用的单例会话未关联到该事务上下文，导致操作不在事务范围内，提交时不会包含这些变更。
• 手动session.close()冗余：with上下文管理器已自动调用close()，手动调用可能导致会话提前关闭，干扰事务正常提交流程。

解决方案 #

方案1：通过request.state传递请求级会话（官方推荐） #

放弃单例模式，利用FastAPI的request.state在请求生命周期内传递会话，天然适配多进程环境：

MiddlewareSession = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    session = MiddlewareSession()
    try:
        request.state.db = session
        response = await call_next(request)
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()
    return response

路由中通过request.state.db获取会话：

@app.get("/items/{item_id}")
async def read_item(request: Request, item_id: int):
    item = request.state.db.query(Item).filter(Item.id == item_id).first()
    return item

方案2：正确使用scoped_session适配多进程 #

若坚持避免依赖注入或request.state，可使用scoped_session为每个请求维护独立会话实例：

from sqlalchemy.orm import scoped_session

# 创建scoped_session实例
SessionLocal = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=engine)
)

@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    try:
        response = await call_next(request)
        SessionLocal.commit()
    except Exception:
        SessionLocal.rollback()
        raise
    finally:
        SessionLocal.remove()
    return response

路由中直接使用SessionLocal即可自动绑定当前请求会话：

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    item = SessionLocal.query(Item).filter(Item.id == item_id).first()
    return item

关键注意事项 #

• 多进程/多线程环境下禁止使用全局单例会话，每个请求必须拥有独立会话，避免交叉污染。
• Gunicorn Worker间内存完全隔离，全局变量无法跨进程共享，必须使用请求级上下文传递会话。
• 事务的提交/回滚必须与请求生命周期绑定，确保请求操作要么全提交，要么全回滚。

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