Executing asynchronous coroutine (asyncio) synchronously in Python

I worked a lot with async functions in languages other than Python. It's usually very straight-forward and easy to find any answers in internet. It was a surprise that solving a very simple question regarding async in Python took me a day. How to execute an async coroutine in a classic synchronous function?

Async library author's position and CPython's GIL made execution of async functions synchronously harder than required. Instead of an exact answer dozens of answers on StackOverflow explain why it is not possible or a bad idea (and a handful of not working suggestions). Ok-ok, I just need a solution!

I had to understand how asyncio works to solve this. So, this code allows to synchronously wait for execution completion of async coroutine without a hot spin-lock:

def sync_exec(coroutine: Awaitable) -> Any:

    try:

        asyncio.get_running_loop()

    except RuntimeError:

        loop = asyncio.new_event_loop()

        return loop.run_until_complete(coroutine)

    result = []

    thread = Thread(target = _execute_in_thread, args = (coroutine, result))

    thread.start()

    thread.join()

    if isinstance(result[0], BaseException):

        raise result[0]

    return result[0]

def _execute_in_thread(coroutine: Awaitable, result: list):

    try:

        loop = asyncio.new_event_loop()

        result.append(loop.run_until_complete(coroutine))

    except BaseException as exc:

        result.append(exc)

Usage:

result1 = sync_exec(_async_func(0.4))

Tests:

async def _async_func(delay):

    await asyncio.sleep(delay)

    await asyncio.sleep(delay)

    return delay

def test_sync_exec():

    delay1 = 0.4

    delay2 = 0.6

   

    result1 = sync_exec(_async_func(delay1))

    assert result1 == delay1

    result2 = sync_exec(_async_func(delay2))

    assert result2 == delay2

@pytest.mark.asyncio

async def test_sync_exec_from_async():

    test_sync_exec()

Explanation:

The idea is to run the async coroutine (or any awaitable) in another thread with its own async loop. It is needed only in case if there is already a running loop in the current thread. Otherwise we could safely(?) execute it in the current thread.

So, first try-except tests exactly this: is there a running loop in the current thread? If get_running_loop() throws it means there is no loop and we are free to create one and execute a function there in a simple manner. Otherwise we spawn a thread with its own loop and execute there. The result (including an exception) is passed from the thread via a mutable container, a dictionary.

Tests include both scenarios: calling from sync function w/o a loop in the thread; and calling from async function with a loop.