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.