1

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?

Sylver11
  • 129
  • 1
  • 8
  • I'm guessing this is more about using the `app.middleware` decorator directly instead of using it to decorate a function; have you tried using `add_middleware` with a class as that's the usual imperative way? – MatsLindh May 09 '22 at 09:30
  • 1
    I couldn't reproduce the exception with the code you provided. If you're using the middleware just for exception handling, I suggest [registering an exception handler](https://fastapi.tiangolo.com/tutorial/handling-errors/#install-custom-exception-handlers) instead of a middleware. – pypae May 09 '22 at 09:39
  • @pypae do you not receive an error when running `pytest`? Regarding the exception handler.. I have implemented it and it works. I guess its also the cleaner solution. Will stick with it thanks ! – Sylver11 May 09 '22 at 10:32
  • Exactly, no error. I'm using python 3.10 and all the newest versions (using `pip install fastapi httpx pytest trio`). Here's the output of the `pytest` command: https://pastebin.com/u3MzGsys – pypae May 09 '22 at 11:11
  • 1
    Please have a look at [this answer](https://stackoverflow.com/a/72833284/17865804), [this answer](https://stackoverflow.com/a/71800464/17865804), as well as [here](https://stackoverflow.com/a/73283272/17865804) and [here](https://stackoverflow.com/a/73464007/17865804). – Chris Jan 22 '23 at 08:07

1 Answers1

0

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.

Sylver11
  • 129
  • 1
  • 8