Run and wait for asynchronous function from a synchronous one using Python asyncio

Question

In my code I have a class with properties, that occasionally need to run asynchronous code. Sometimes I need to access the property from asynchronous function, sometimes from synchronous - that's why I don't want my properties to be asynchronous. Besides, I have an impression that asynchronous properties in general is a code smell. Correct me if I'm wrong.

I have a problem with executing the asynchronous method from the synchronous property and blocking the further execution until the asynchronous method will finish.

Here is a sample code:

import asyncio


async def main():
    print('entering main')
    synchronous_property()
    print('exiting main')


def synchronous_property():
    print('entering synchronous_property')
    loop = asyncio.get_event_loop()
    try:
        # this will raise an exception, so I catch it and ignore
        loop.run_until_complete(asynchronous())
    except RuntimeError:
        pass
    print('exiting synchronous_property')


async def asynchronous():
    print('entering asynchronous')
    print('exiting asynchronous')


asyncio.run(main())

Its output:

entering main
entering synchronous_property
exiting synchronous_property
exiting main
entering asynchronous
exiting asynchronous

First, the RuntimeError capturing seems wrong, but if I won't do that, I'll get RuntimeError: This event loop is already running exception.

Second, the asynchronous() function is executed last, after the synchronous one finish. I want to do some processing on the data set by asynchronous method so I need to wait for it to finish. If I'll add await asyncio.sleep(0) after calling synchronous_property(), it will call asynchronous() before main() finish, but it doesn't help me. I need to run asynchronous() before synchronous_property() finish.

What am I missing? I'm running python 3.7.

Answer 1

Asyncio is really insistent on not allowing nested loops, by design. However, you can always run another event loop in a different thread. Here is a variant that uses a thread pool to avoid having to create a new thread each time around:

import asyncio, concurrent.futures

async def main():
    print('entering main')
    synchronous_property()
    print('exiting main')

pool = concurrent.futures.ThreadPoolExecutor()

def synchronous_property():
    print('entering synchronous_property')
    result = pool.submit(asyncio.run, asynchronous()).result()
    print('exiting synchronous_property', result)

async def asynchronous():
    print('entering asynchronous')
    await asyncio.sleep(1)
    print('exiting asynchronous')
    return 42

asyncio.run(main())

This code creates a new event loop on each sync->async boundary, so don't expect high performance if you're doing that a lot. It could be improved by creating only one event loop per thread using asyncio.new_event_loop, and caching it in a thread-local variable.

Answer 2

The easiest way is using an existing "wheel", like asgiref.async_to_sync

from asgiref.sync import async_to_sync

then:

async_to_sync(main)()

in general:

async_to_sync(<your_async_func>)(<.. arguments for async function ..>)

This is a caller class which turns an awaitable that only works on the thread with the event loop into a synchronous callable that works in a subthread.

If the call stack contains an async loop, the code runs there. Otherwise, the code runs in a new loop in a new thread.

Either way, this thread then pauses and waits to run any thread_sensitive code called from further down the call stack using SyncToAsync, before finally exiting once the async task returns.

Answer 3

There appears to a problem with the question as stated. Restating the question: How to communicate between a thread (containing no async processes and hence considered sync) and an async proces (running in some event loop). One approach is to use two sync Queues. The sync process puts its request/parameters into the QtoAsync, and waits on the QtoSync. The async process reads the QtoAsync WITHOUT wait, and if it finds a request/parameters, executes the request, and places the result in QtoSync.

import queue
QtoAsync = queue.Queue()
QtoSync = queue.Queue()
...

async def asyncProc():
    while True:
        try:
            data=QtoAsync.get_nowait()
            result = await <the async that you wish to execute>
            QtoAsync.put(result) #This can block if queue is full. you can use put_nowait and handle the exception.
        except queue.Empty:
            await asyncio.sleep(0.001) #put a nominal delay forcing this to wait in event loop
....
#start the sync process in a different thread here..
asyncio.run(main()) #main invokes the async tasks including the asyncProc

The sync thread puts it request to async using:
req = <the async that you wish to execute>
QtoAsync.put(req)
result = QtoSync.get()

This should work.

Problem with the question as stated: 1. When the async processes are started with asyncio.run (or similar) execution blocks until the async processes are completed. A separate sync thread has to be started explicity before calling asyncio.run 2. In general asyncio processes depend on other asyncio processes in that loop. So calling a async process from another thread is not permitted directly. The interaction should be with the event loop, and using two queues is one approach.

Answer 4

I want to make the async call to execute from sync and block it's execution

Just make the sync func async and await the asynchronous function. Async functions are just like normal functions and you can put whatever code you want in them. If you still have a problem modify your question using actual code you are trying to run.

import asyncio


async def main():
    print('entering main')
    await synchronous_property()
    print('exiting main')


async def synchronous_property():
    print('entering synchronous_property')

    await asynchronous()

    # Do whatever sync stuff you want who cares

    print('exiting synchronous_property')


async def asynchronous():
    print('entering asynchronous')
    print('exiting asynchronous')


asyncio.run(main())