sqlalchemypytestfastapiasyncpgpytest-asyncio

FastAPI SQLAlchemy Async tests fail without poolclass set to NullPool


I am building an app using FastAPI and SQLAlchemy. Async setup. For some reason I cannot find a proper way to setup async tests. I am running a very basic test setup.

conftest.py:


@pytest_asyncio.fixture(scope='function')
async def client() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

test_event.py:

@pytest.mark.asyncio
async def test_get_public_events(client: AsyncClient) -> None:
    response = await client.get("/api/v1/events")
    assert response.status_code == 200

The interesting thing is that tests only work if I set poolclass=NullPool on my async_engine.

db.py:

async_engine = create_async_engine(
    url=settings.POSTGRES_URL.get_secret_value(),
    poolclass=NullPool, # TODO: Research this, tests are failing if this is not set
    pool_pre_ping=True,
    echo=True, # TODO: Handle this in a better way

)

async_session = async_sessionmaker(
    bind=async_engine,
    autoflush=False,
    expire_on_commit=False,
    class_=AsyncSession,
)

If I dont then I get an error when running tests:

RuntimeError: Task <Task pending name='Task-4'
 coro=<test_get_public_events() running at ....
 cb=[_run_until_complete_cb() at ...... got
 Future <Future pending> attached to a different loop

Versions:

fastapi = "^0.110.0"
sqlalchemy = {extras = ["asyncio"], version = "^2.0.28"}
asyncpg = "^0.29.0"
pytest = "^8.1.1"

Maybe there is someone more knowledgable about the issues and can explain what is going on here?

Thanks


Solution

  • Managed to fix the issue by adding await async_engine.dispose() to fixture which prepares my database for testing:

    @pytest_asyncio.fixture(scope="session", autouse=True)
    async def prepare_database() -> AsyncIterator[None]:
    
        # Creates tables
        async with async_engine.begin() as conn:
            await conn.run_sync(CustomBaseModel.metadata.create_all)
    
        # Creates test user
        async with async_session_factory.begin() as session:
            session.add(UserModel(...))
            await session.commit()
    
        await async_engine.dispose() # Added this line
    
        yield