FastAPI Async Test Failing With Middleware

1.6k views Asked by At

I want to add middleware to my FastAPI instance in order to do some custom exception handling (logging etc) but once added I receive the following error on my pytest endpoint healthcheck:

    self = <asyncio.unix_events._UnixDefaultEventLoopPolicy object at 0x1052ece20>

    def get_event_loop(self):
        """Get the event loop for the current context.

        Returns an instance of EventLoop or raises an exception.
        """
        if (self._local._loop is None and
                not self._local._set_called and
                threading.current_thread() is threading.main_thread()):
            self.set_event_loop(self.new_event_loop())

        if self._local._loop is None:
>           raise RuntimeError('There is no current event loop in thread %r.'
                               % threading.current_thread().name)
E           RuntimeError: There is no current event loop in thread 'MainThread'.

Minimal reproducible example:

main.py

from fastapi import FastAPI, Request


async def request_handler(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as ex:
        # Do something specific here 
        raise ex

   
app = FastAPI()
app.middleware("http")(request_handler)


@app.get("/healthcheck")
async def healthcheck():
    return {"status": "ok"}

test_main.py

import pytest    
from httpx import AsyncClient
from .main import app

@pytest.mark.anyio
async def test_root():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/healthcheck")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}

Question: Is there a solution or a work around to this problem?

1

There are 1 answers

0
Sylver11 On

To answer my own question. I ended up doing the following:

@app.exception_handler(Exception)
async def exception_handler(request: Request, exc: Exception):
    capture_exception(exc)
    return exc.response()

Where capture_exception() comes from Sentry.

On another note, I did end up completely removing the async test suite and just replaced it with the normal setup as described in the FastApi docs. That also works and did not cause any problems with middleware.