47

I am using asyncio for an application in a very basic way. Checking most tutorials around the internet (and even the official docs), I see that they use the get_event_loop() and loop.run_until_complete():

import asyncio async def say(what, when): await asyncio.sleep(when) print(what) loop = asyncio.get_event_loop() loop.run_until_complete(say('hello world', 1)) loop.close()

But in the Python 3.7 docs, we can read:

Application developers should typically use the high-level asyncio functions, such as asyncio.run(), and should rarely need to reference the loop object or call its methods. This section is intended mostly for authors of lower-level code, libraries, and frameworks, who need finer control over the event loop behavior.

I found it much cleaner and easier to use, but it only works for Python 3.7+. So here I would have to make a choice, whether to use Python 3.7+ and run() or make it compatible with Python 3.6 and use the event loop. How would you manage this? Is there a simple way to make it backwards compatible with Python 3.6? Should I check Python version first and use either one way or another based on that, until Python 3.7 becomes a common version?

Improve this question
3
  • 2
    If you're going to write more complicated code that works on both versions, and simpler code for newer versions, and you'll switch dynamically between them… isn't it easier to simply stick with the more complicated one that works on both? Commented Apr 9, 2019 at 10:14
  • 1
    @deceze yes, maybe that's the best option, I wanted to have an opinion on that and in the case of making it compatible, to know which is the best way to do so Commented Apr 9, 2019 at 10:27
  • 1
    @deceze Emulating asyncio.run on older Python versions is not hard, and you get the advantage that you test your code under the conditions set up by asyncio.run, i.e. on a freshly created event loop. Commented Apr 9, 2019 at 14:57

2 Answers 2

Reset to default
38

Is there a simple way to make [code making use of asyncio.run] backwards compatible with Python 3.6?

You can implement a simple substitute for asyncio.run and call it on older Python versions:

import asyncio, sys, types def run(coro): if sys.version_info >= (3, 7): return asyncio.run(coro) # Emulate asyncio.run() on older versions # asyncio.run() requires a coroutine, so require it here as well if not isinstance(coro, types.CoroutineType): raise TypeError("run() requires a coroutine object") loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: return loop.run_until_complete(coro) finally: loop.close() asyncio.set_event_loop(None)

The advantage of this approach over just using loop.run_until_complete() is that you're executing your code under the semantics close to those of the new asyncio.run, even on older Python versions. (For example, you will always run on a freshly created event loop.) Dropping support for pre-3.7 Python will be as easy as removing the run shim and calling asyncio.run directly.

Improve this answer
Sign up to request clarification or add additional context in comments.

4 Comments

Nice! This looks great! I will try this :) Thanks!
just a minor thing on your code, the condition should be sys.version_info >= (3, 7). Thanks for the help! :)
Suspect the order of the 2 lines in the finally block should not be the opposite of that in the Python standard library (the other answer on this question)
@LouisMaddox I don't think it matters, but if in doubt, feel free to reverse the order
4

It's possible to replicate asyncio.run by copying the code from the asyncio.runners.py. The one below is from Python 3.8.

from asyncio import coroutines, events, tasks def run(main, *, debug=False): """Execute the coroutine and return the result. This function runs the passed coroutine, taking care of managing the asyncio event loop and finalizing asynchronous generators. This function cannot be called when another asyncio event loop is running in the same thread. If debug is True, the event loop will be run in debug mode. This function always creates a new event loop and closes it at the end. It should be used as a main entry point for asyncio programs, and should ideally only be called once. Example: async def main(): await asyncio.sleep(1) print('hello') asyncio.run(main()) """ if events._get_running_loop() is not None: raise RuntimeError( "asyncio.run() cannot be called from a running event loop") if not coroutines.iscoroutine(main): raise ValueError("a coroutine was expected, got {!r}".format(main)) loop = events.new_event_loop() try: events.set_event_loop(loop) loop.set_debug(debug) return loop.run_until_complete(main) finally: try: _cancel_all_tasks(loop) loop.run_until_complete(loop.shutdown_asyncgens()) finally: events.set_event_loop(None) loop.close() def _cancel_all_tasks(loop): to_cancel = tasks.all_tasks(loop) if not to_cancel: return for task in to_cancel: task.cancel() loop.run_until_complete( tasks.gather(*to_cancel, loop=loop, return_exceptions=True)) for task in to_cancel: if task.cancelled(): continue if task.exception() is not None: loop.call_exception_handler({ 'message': 'unhandled exception during asyncio.run() shutdown', 'exception': task.exception(), 'task': task, })
Improve this answer

1 Comment

In version 3.9 they added loop.run_until_complete(loop.shutdown_default_executor()) after the shutdown_asyncgens (otherwise identical), and no change in 3.10

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.