First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import pytest
import sqlalchemy as sa

from asyncio import current_task
from fastapi import APIRouter
from fastapi import Depends
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import async_scoped_session
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.declarative import DeclarativeMeta
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker

DB_URL = "SOME_DB_URL"
Base: DeclarativeMeta = declarative_base()
app = FastAPI()
test_router = APIRouter(prefix="test")
app.include_router(test_router)


class User(Base):

    __tablename__ = "users"
    user_id = sa.Column(sa.String, primary_key=True)
    name = sa.Column(sa.String)
    

def get_db_session() -> async_scoped_session:

    engine = create_async_engine(DB_URL)
    session_maker = sessionmaker(bind=engine, class_=AsyncSession)
    db_session = async_scoped_session(session_maker, scopefunc=current_task)

    return db_session


@test_router.post(path="/")
async def hello(
    user_id: str,
    db_session: async_scoped_session = Depends(get_db_session)
) -> str:
    
    query = select(User).where(User.user_id == user_id)
    result = await db_session.execute(query)
    user_name = result.scalars().first().name

    return f"hello {user_name}"


@pytest.fixture
def test_client():

    # some dependency overrides here, omitted for clarity as they don't impact the issue
    # app.dependency_overrides[get_db_session] = some_method_to_create_test_db_session
    
    yield TestClient(app)

    # reset overrides
    app.dependency_overrides = {}


async def test_post_success(test_client: TestClient):

    response = test_client.post(url="test/", params={"user_id": "some_id"})

    assert response == "hello some_user"

Description

Hi everyone,

I need some help regarding the best way to test a FastAPI app interacting with a postgresql databse through asyncpg. More precisely, I am trying to swap psycopg2 for asyncpg, and rewriting all necessary methods and unit tests to make them async.

I'm using pytest-asyncio to unit test the newly-async code. Testing calls to the API is proving problematic: all async tests now fail with the dreaded Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop.

From what I could find on pytest-asyncio's github, this kind of issue is not easily solvable on their end, and is usually addressed by creating an event_loop fixture and passing it to the app being tested with something like:

@pytest.fixture(scope='session')
def event_loop():
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    loop.close()

However, this type of fix was suggested in cases that don't involve FastAPI/Uvicorn, and I am at a loss as to how to force-set FastAPI's event loop to use the test fixture.

Could anyone suggest the best way to do so, or an alternative method to solve this testing issue?

I've tried to submit a code snipped encapsulating the main details of my use case: the router structure, the db session generation and usage, and the test client fixture. Please feel free to request a rewrite of the example or additional info as needed.

Thanks in advance for your help

Operating System

Linux

Operating System Details

Debian

FastAPI Version

fastapi==0.85.0

Python Version

Python 3.9.2

Additional Context

No response