Python: Call asynchronous code from synchronous method when there is already an event loop running [duplicate]

Question

I am working with FastAPI and uvloop to serve a REST API in an efficient way.

I have a lot of asynchronous code that make calls to remote resources such as a database, a storage, etc, those functions looks like this:

async def _get_remote_resource(key: str) -> Resource:
    # do some async work
    return resource

I'm implementing an interface to an existing Abstract Base Class where I need to use the asynchronous function from above in a synchronous method. I have done something like:

class Resource:
     def __str__(self):
         resource = asyncio.run_until_complete(_get_remote_resource(self.key))
         return f"{resource.pk}"

Great! Now I do an endpoint in fastapi to make this work accesible:

@app.get("")
async def get(key):
     return str(Resource(key))

The problem is that FastAPI already gets and event loop running, using uvloop, and then the asynchronous code fails because the loop is already running.

Is there any way I can call the asynchronous method from the synchronous method in the class? Or do I have to rethink the structure of the code?

Answer 1

The runtime error is designed precisely to prevent what you are trying to do. run_until_complete is a blocking call, and using it inside an async def will halt the outer event loop.

The straightforward fix is to expose the needed functionality through an actual async method, e.g.:

class Resource:
    def name(self):
        return loop.run_until_complete(self.name_async())

    async def name_async(self):
        resource = await _get_remote_resource(self.key)
        return f"{resource.pk}"

Then in fastapi you'd access the API in the native way:

@app.get("")
async def get(key):
     return await Resource(key).name_async()

You could also define __str__(self) to return self.name(), but that's best avoided because something as basic as str() should be callable from within asyncio as well (due to use in logging, debugging, etc.).

Answer 2

I would like to complement the @user4815162342 answer.

FastAPI is an asychronous framework. I would suggest sticking to a few principles:

• Do not execute IO operations in synchronous functions in a blocking way. Prepare this resource asynchronously and already pass the ready data to the synchronous function (this principle can be called an asynchronous dependency for synchronous code).
• If you still need to perform a blocking IO operation in a synchronous code, then do it in a separate thread. And wait for this result asynchronously by means of asyncio (def endpoint, run_in_executor with ThreadPoolExecutor or def background task).
• If you need to do a blocking CPU-bound operation, then delegate its execution to a separate process (the simplest way run_in_executor with ProcessPoolExecutor or any task queue).