FastAPI+SQLAlchemy2.0异步测试报错：任务与Future绑定不同事件循环

核心原因 #

你的POST接口触发的RuntimeError是因为scoped_session的事件循环绑定逻辑错误：视图依赖的scoped_session_dependency在测试环境中没有绑定到测试用的事件循环，导致CRUD操作的AsyncSession和请求处理的事件循环不一致，进而引发Future绑定错误。同时会话未正确归还连接池，触发连接泄漏警告。GET接口正常是因为其逻辑未暴露跨事件循环的会话问题。

解决方案 #

1. 修正Scoped Session的异步上下文绑定 #

修改db_helper.py中的scoped_session_dependency实现，使用**上下文变量（ContextVar）**替代线程本地存储，适配异步环境的作用域管理：

# db_helper.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from contextvars import ContextVar

# 用上下文变量绑定会话作用域，适配异步事件循环
session_scope_var: ContextVar[str] = ContextVar("session_scope")

async def scoped_session_dependency() -> AsyncSession:
    # 初始化异步引擎和会话工厂
    engine = create_async_engine(
        "your_database_url",
        pool_pre_ping=True,  # 防止连接池失效
        pool_recycle=300
    )
    async_session_factory = sessionmaker(
        bind=engine,
        class_=AsyncSession,
        expire_on_commit=False
    )
    # 基于上下文变量创建scoped session
    scoped_session_instance = scoped_session(
        async_session_factory,
        scopefunc=session_scope_var.get
    )
    
    try:
        session = scoped_session_instance()
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise
    finally:
        await scoped_session_instance.remove()  # 归还连接到池

2. 统一测试与视图的会话依赖 #

在conftest.py中覆盖依赖，确保测试客户端使用和视图一致的会话上下文：

# conftest.py
import pytest
from fastapi.testclient import TestClient
from your_app.main import app
from your_app.db_helper import scoped_session_dependency

@pytest.fixture(scope="function")
def client():
    # 覆盖会话依赖，确保测试使用同一上下文
    async def override_scoped_session():
        yield from scoped_session_dependency()
    
    app.dependency_overrides[scoped_session_dependency] = override_scoped_session
    yield TestClient(app)
    # 清理依赖和会话
    app.dependency_overrides.clear()

3. 规范POST接口测试逻辑 #

测试代码中直接使用测试客户端发起请求，不要手动创建会话，确保依赖注入的会话正确绑定到测试事件循环：

# test_menu.py
def test_create_menu(client):
    test_menu = {"name": "Test Main Menu", "description": "Sample menu"}
    response = client.post("/menus/", json=test_menu)
    
    assert response.status_code == 201
    response_data = response.json()
    assert response_data["name"] == test_menu["name"]
    assert response_data["description"] == test_menu["description"]

4. 修复连接池泄漏问题 #

确保db_helper.py的finally块中执行await scoped_session_instance.remove()，这会强制将连接归还到连接池。同时在测试fixture的清理阶段，确保所有会话都被正确销毁。

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