FastAPI middleware performance

 As per the FastAPI docs, the way to create and add custom middlewares is

@app.middleware("http")

async def add_my_middlware(request: Request, call_next):

response = await call_next(request)

return response

Seems simple enough. Before the await, you can do something with the request. After the await, you can do something with the response.

But if you run benchmark, you will find something very surprising.

Python - calling async function from a sync code flow

 Recently I ran into a scenario where I needed to call a init function for a third party library in my FastAPI application. The problem was, the function was async.

One way of calling an async function from a sync flow is using asyncio event loop.

import asyncio

loop = asyncio.get_event_loop()

loop.run_until_complete(init())

But this gave event loop is already running error.

The problem is, FastAPI application is run with uvicorn which starts a loop.

So I tried creating a new loop.

loop = asyncio.new_event_loop()

asyncio.set_event_loop(loop)

loop.run_until_complete(init())

But this still didn't work as asyncio only supports one loop at a time.

The most suggested approach is to use nest-asyncio. 

import nest_asyncio

loop = asyncio.new_event_loop()

nest_asyncio.apply(loop)

asyncio.set_event_loop(loop)

loop.run_until_complete(init())

This raised an exception: Can't patch uvloop.Loop.

uvicorn patches asyncio to use uvloop which is better performant that vanilla asyncio (previous claims were 2x to 4x. In the simple test I ran, even with performance changes in 3.12, asyncio was about 25% slower than uvloop.).

Did some research, but couldn't find any solution around this other than forcing uvicorn to run with vanilla asyncio

uvicorn main:app --loop asyncio

It didn't make sense to take a performance hit, just to call a function.

So I decided to dig deeper into why asyncio.new_event_loop returns uvloop.Loop.

The way uvloop does it is, it sets the asyncio event_loop_policy.

This gave me an idea, what if we temporarily restore the event loop policy, get a loop, apply nest_asyncio and then restore the event loop policy.

import nest_asyncio

import asyncio

_cur_event_loop_policy = asyncio.get_event_loop_policy()

asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())

loop = asyncio.new_event_loop()

nest_asyncio.apply(loop) # type: ignore

asyncio.set_event_loop(loop)

result = loop.run_until_complete(init())

loop.close()

asyncio.set_event_loop_policy(_cur_event_loop_policy)

This seems to do the trick and I am able to call an async function in my FastAPI application main.py before initializing app.