Calling coroutines from sync code

Python Help
,
1

Suppose I want to call async code from a regular function and wait for its return. This function could be part of a library or could be far down a stack that is being iteratively converted to asyncio, so the fact that it’s using asyncio internally should be transparent to any callers. Maybe it’s called from async contexts and there already is an event loop, maybe there isn’t. What’s currently the best way to call a coroutine in these contexts?

async def fetch_something(): ...

def sync_func():  # could be called from async or sync contexts
    ret = call_async(fetch_something())
    ...

Options I have considered for call_async:

  • asyncio.run() – This is wrong, according to the documentation: “This function cannot be called when another asyncio event loop is running in the same thread. […] It should be used as a main entry point for asyncio programs, and should ideally only be called once.”
  • asyncio.get_event_loop().run_until_complete() – This has started printing warnings when called from a synchronous context.
  • asyncio.new_event_loop().run_until_complete()?
  • Write a custom utility that calls get_running_loop() and new_event_loop() if that throws an exception?

Also, how can I handle the various housekeeping that asyncio.run() does?

4 Likes
2

Another solution might be a custom function like this:

def call_async(coro):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    else:
        return loop.run_until_complete(coro)

This should take care of all necessary cleanup in case we’re not running in an event loop. Would that make sense?

3

I tend to do something similar to your call_async(). Though I’m also not sure if its the ‘right way’.

4

This works but why make everyone repeat this code?

why not have loop.run_until_complete create a loop if necessary?

5

You can make main() asynchronous, and then you can schedule a coroutine using asyncio.create_task(asynchronous()).
You can continue using the synchronous functions as before.

import asyncio
import time


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

    synchronous()

    await asyncio.sleep(0.0)


def synchronous():
    print('synchronous()')
    time.sleep(1.0)

    asyncio.create_task(asynchronous())


async def main():
    synchronous()

    await asyncio.sleep(0.0)


def run():
    loop = asyncio.new_event_loop()
    try:
        loop.run_until_complete(main())
        loop.run_forever()

    finally:
        loop.close()


run()
1 Like