FastAPI异步测试遇RuntimeError：任务绑定不同事件循环

我用pytest测试FastAPI接口，采用SQLAlchemy异步连接PostgreSQL（依赖asyncpg驱动）。运行测试时出现RuntimeError，提示任务获取了绑定到不同事件循环的Future，其中test_update_client测试用例失败，test_create_client通过。推测是数据库操作使用了另一个事件循环，希望在不Mock数据库CRUD逻辑的前提下解决该问题。

相关代码 #

main.py #

from fastapi import FastAPI, status, HTTPException, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from .schema import ClientIn, ClientOut, ClientInWithID, Client, ValidationErrorSchema
from . import clients
from .database import SessionLocal


app = FastAPI()


async def get_db() -> AsyncSession:
    if hasattr(get_db, "db"):
        db: AsyncSession = get_db.db
        return db
    db = SessionLocal()
    setattr(get_db, "db", db)
    return db


@app.post("/client/",
          response_model=ClientOut,
          tags=["client"],
          responses={422: {"model": ValidationErrorSchema}}
          )
async def create_client(client_in: ClientIn, db: AsyncSession = Depends(get_db)) -> Client:
    client = await clients.create_client(db, client_in)
    return client


@app.put("/client/",
         response_model=ClientOut | None,
         tags=["client"],
         responses={422: {"model": ValidationErrorSchema}, 404: {}}
         )
async def update_client(client: ClientInWithID, db: AsyncSession = Depends(get_db)) -> Client | None:
    db_client = await clients.get_client_by_id(db, client.id)
    if not db_client:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    return await clients.update_client(db, client)

test_main.py #

import pytest

from httpx import AsyncClient

from app import main


@pytest.mark.asyncio
async def test_create_client():
    data = {
            "phone_number": "+79009999999",
            "phone_operator_code": 900,
            "timezone": "Europe/Amsterdam",
            "tag": {
                "text": "Any text"
            }
        }

    async with AsyncClient(app=main.app, base_url="http://localhost:8000") as client:
        response = await client.post(url="client/", json=data)

    assert response.status_code == 200


@pytest.mark.asyncio
async def test_update_client():
    data = {
            "id": 1,
            "phone_number": "+79009900000",
            "phone_operator_code": 900,
            "timezone": "Europe/Amsterdam",
            "tag": {
                "text": "Fuck this shit"
            }
        }

    async with AsyncClient(app=main.app, base_url="http://localhost:8000") as client:
        response = await client.put(url="client/", json=data)

    assert response.status_code == 200

报错信息 #

venv/lib/python3.11/site-packages/asyncpg/connection.py:565: in prepare
    return await self._prepare(
venv/lib/python3.11/site-packages/asyncpg/connection.py:583: in _prepare
    stmt = await self._get_statement(
venv/lib/python3.11/site-packages/asyncpg/connection.py:397: in _get_statement
    statement = await self._protocol.prepare(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

>   ???
E   RuntimeError: Task <Task pending name='Task-3' coro=<test_update_client() running at /home/kryseyt/code/python/BackendTask1/tests/test_main.py:38> cb=[_run_until_complete_cb() at /home/kryseyt/.python3.11/lib/python3.11/asyncio/base_events.py:180]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop

asyncpg/protocol/protocol.pyx:168: RuntimeError
================================================ short test summary info =========================================
FAILED tests/test_main.py::test_update_client - RuntimeError: Task <Task pending name='Task-3' coro=<test_update_client() running at /home/kryseyt/code/python/BackendTask1/tests/test_main.py:38> cb=[_run_until_complet...
======================================================================== 1 failed, 1 passed in 5.82s =========================================================================

解决方案 #

问题根源在于get_db函数把数据库会话绑定到函数属性get_db.db实现全局复用，但每个pytest异步测试用例会启动独立的事件循环，第一个测试用例创建的会话绑定在旧循环上，第二个测试用例复用该会话时，就会出现Future绑定到不同事件循环的错误。

步骤1：修复数据库会话管理 #

修改main.py中的get_db函数，确保每个请求使用独立会话并自动关闭：

async def get_db() -> AsyncSession:
    async with SessionLocal() as db:
        yield db

这样每次调用get_db都会生成新的会话，请求结束后自动关闭，避免跨测试用例复用会话导致的循环冲突。

步骤2：添加测试用例的数据库隔离 #

为保证测试用例之间状态互不干扰，可通过fixture创建独立会话并在测试后回滚事务：

import pytest
from httpx import AsyncClient
from app import main
from app.database import SessionLocal
from sqlalchemy.ext.asyncio import AsyncSession


@pytest.fixture(scope="function")
async def db_session():
    async with SessionLocal() as session:
        async with session.begin():
            yield session
        await session.rollback()  # 测试结束后回滚事务，隔离状态


@pytest.mark.asyncio
async def test_create_client(db_session):
    # 替换app的数据库依赖为测试会话
    def override_get_db():
        yield db_session

    main.app.dependency_overrides[main.get_db] = override_get_db

    data = {
            "phone_number": "+79009999999",
            "phone_operator_code": 900,
            "timezone": "Europe/Amsterdam",
            "tag": {
                "text": "Any text"
            }
        }

    async with AsyncClient(app=main.app, base_url="http://localhost:8000") as client:
        response = await client.post(url="client/", json=data)

    assert response.status_code == 200
    main.app.dependency_overrides.clear()


@pytest.mark.asyncio
async def test_update_client(db_session):
    def override_get_db():
        yield db_session

    main.app.dependency_overrides[main.get_db] = override_get_db

    # 先创建测试数据，避免update时找不到id
    create_data = {
            "phone_number": "+79009999999",
            "phone_operator_code": 900,
            "timezone": "Europe/Amsterdam",
            "tag": {
                "text": "Any text"
            }
        }
    async with AsyncClient(app=main.app, base_url="http://localhost:8000") as client:
        create_resp = await client.post(url="client/", json=create_data)
        assert create_resp.status_code == 200
        client_id = create_resp.json()["id"]

        update_data = {
                "id": client_id,
                "phone_number": "+79009900000",
                "phone_operator_code": 900,
                "timezone": "Europe/Amsterdam",
                "tag": {
                    "text": "Updated text"
                }
            }
        response = await client.put(url="client/", json=update_data)

    assert response.status_code == 200
    main.app.dependency_overrides.clear()

这样修改后，每个测试用例使用独立的数据库会话和事务，既解决了事件循环冲突，又保证了测试用例之间的状态隔离，同时无需Mock数据库CRUD逻辑，真实验证数据库操作。

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