Fastapi 测试 RuntimeError：任务附加到不同的循环

Question

我尝试用 pytest 测试我的端点

主要.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)

测试_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

我使用 sqlalchemy 及其使用 asyncpg 连接到 postgres，因为 asyncpg 我有错误：

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 =========================================================================

发生这种情况是因为正在创建另一个事件循环来与数据库一起使用，但我能用它做什么呢？
我可以在不嘲笑我的数据库 CRUD 的情况下对此做点什么吗？

Answer 1

尝试在引擎构造函数中添加“poolclass=NullPool”参数。

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host/dbname",
    poolclass=NullPool,
)

资料来源：

• https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#using-multiple-asyncio-event-loops
• RuntimeError：任务 ___ 在 ___ 处运行，Future <Futureending cb=[Protocol._on_waiter_completed()]> 附加到不同的循环