FastAPI中使用SQLAlchemy异步会话触发事务已开启错误

场景复现 #

我在FastAPI应用中通过中间件将SQLAlchemy异步会话绑定到request.state，代码如下：

from fastapi import FastAPI, Request
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession, create_async_engine
from starlette.middleware.base import BaseHTTPMiddleware

# 创建异步引擎
engine = create_async_engine("<my postgres url>", future=True, pool_pre_ping=True, echo=True)

# 创建异步会话工厂
AsyncSessionFactory = async_sessionmaker(
    bind=engine, 
    autoflush=False, 
    expire_on_commit=False, 
    future=True, 
    class_=AsyncSession
)

# 自定义中间件：将会话绑定到request.state
class AddDBSessionToRequest(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        async with AsyncSessionFactory() as db_session:
            try:
                request.state.db_session = db_session
                print(f"Middleware transaction status: {db_session.in_transaction()}")  # 输出False
                response = await call_next(request)
            except SQLAlchemyError:
                logger.exception("SQLAlchemy error when processing '%s %s'", request.method, request.url)
                response = Response("Internal server error", status_code=500)
        return response

# 中间件列表
middleware = [
    Middleware(
        CORSMiddleware,
        allow_origins=get_settings().cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    ),
    Middleware(TrustedHostMiddleware, allowed_hosts=get_settings().allowed_hosts),
    Middleware(MeasureRequestProcessingTime),
    Middleware(AddDBSessionToRequest),
]

app = FastAPI(middleware=middleware)

# 测试视图
@app.post("/test")
async def test_view(request: Request):
    db_session = request.state.db_session
    print(f"view transaction status: {db_session.in_transaction()}")  # 输出True
    async with db_session.begin():
        # 执行数据库操作
        pass

运行后触发错误：

sqlalchemy.exc.InvalidRequestError: A transaction is already begun on this Session.

完整堆栈跟踪：

Traceback (most recent call last):
  File "/opt/venv/lib/python3.11/site-packages/anyio/streams/memory.py", line 94, in receive
    return self.receive_nowait()
           ^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/anyio/streams/memory.py", line 89, in receive_nowait
    raise WouldBlock
anyio.WouldBlock

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/venv/lib/python3.11/site-packages/starlette/middleware/base.py", line 78, in call_next
    message = await recv_stream.receive()
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/anyio/streams/memory.py", line 114, in receive
    raise EndOfStream
anyio.EndOfStream

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/test/app.py", in dispatch
    response = await call_next(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/starlette/middleware/base.py", line 84, in call_next
    raise app_exc
  File "/opt/venv/lib/python3.11/site-packages/starlette/middleware/base.py", line 70, in coro
    await self.app(scope, receive_or_disconnect, send_no_error)
  File "/opt/venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
    raise exc
  File "/opt/venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
    await self.app(scope, receive, sender)
  File "/opt/venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 21, in __call__
    raise e
  File "/opt/venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__
    await self.app(scope, receive, send)
  File "/opt/venv/lib/python3.11/site-packages/starlette/routing.py", line 718, in __call__
    await route.handle(scope, receive, send)
  File "/opt/venv/lib/python3.11/site-packages/starlette/routing.py", line 276, in handle
    await self.app(scope, receive, send)
  File "/opt/venv/lib/python3.11/site-packages/starlette/routing.py", line 66, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/fastapi/routing.py", line 237, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/fastapi/routing.py", line 163, in run_endpoint_function
    return await dependant.call(**values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/test/app.py", in test_view
    async with db_session.begin():
  File "/opt/venv/lib/python3.11/site-packages/sqlalchemy/ext/asyncio/base.py", line 127, in __aenter__
    return await self.start(is_ctxmanager=True)  # type: ignore
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/sqlalchemy/ext/asyncio/session.py", line 1769, in start
    await greenlet_spawn(
  File "/opt/venv/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 179, in greenlet_spawn
    result = context.switch(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1811, in begin
    raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: A transaction is already begun on this Session.

对比测试 #

我编写了纯异步测试代码，未出现该错误：

import asyncio

async def another_test():
    async with AsyncSessionFactory() as db_session:
        async with db_session.begin():
            # 执行数据库操作
            pass

loop = asyncio.get_event_loop()
loop.run_until_complete(another_test())

原因分析 #

核心差异在于FastAPI的请求处理流程与纯异步代码的执行逻辑：

1. SQLAlchemy异步会话默认开启autobegin=True，当会话执行第一个需要数据库连接的操作时，会自动启动事务。
2. 在FastAPI场景中，从中间件将db_session存入request.state到视图获取它的这段时间内，存在隐式的数据库操作（可能来自其他中间件、框架内部逻辑，或是未注意到的代码）触发了autobegin，导致事务已经处于活跃状态，此时再调用db_session.begin()就会触发冲突报错。
3. 纯异步测试代码中，会话创建后直接手动启动事务，中间没有任何触发autobegin的操作，因此不会出现冲突。

解决方案 #

你提到的使用begin_nested()是可行的，它会在已有的活跃事务中创建嵌套事务（即数据库保存点）。另外还有两种常见处理方式：

• 手动关闭autobegin：在会话工厂初始化时设置autobegin=False，完全由代码手动控制事务的启动与提交：

  AsyncSessionFactory = async_sessionmaker(
      bind=engine, 
      autoflush=False, 
      expire_on_commit=False, 
      future=True, 
      class_=AsyncSession,
      autobegin=False  # 关闭自动事务
  )
  

• 中间件内启动事务：在中间件中直接启动事务，视图复用该事务即可，无需再手动调用begin()：

  class AddDBSessionToRequest(BaseHTTPMiddleware):
      async def dispatch(self, request, call_next):
          async with AsyncSessionFactory() as db_session:
              async with db_session.begin():
                  try:
                      request.state.db_session = db_session
                      response = await call_next(request)
                  except SQLAlchemyError:
                      logger.exception("SQLAlchemy error when processing '%s %s'", request.method, request.url)
                      response = Response("Internal server error", status_code=500)
          return response
  

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