Python asyncio.run boilerplate for Python >= 3.5

Concurrency is builtin to Python via asyncio. Python 3.6 added AsyncIO Generators, which greatly streamline asynchronous algorithms best expressed with generators. The async for introduced in Python 3.5 also simplifies expression of asynchronous for loops.

As in Julia, the expression of asynchronous structures in Python does not implement concurrent execution. Concurrent execution in Python is governed by collections of tasks or futures such as asyncio.gather and initiated by a runner such as asyncio.run

However, asyncio.run doesn’t currently handle all use cases across operating systems. AsyncIO subprocess in particular needs special options when using Python 3.6 or 3.7. The options needed are not the same for every project, depending on the asynchronous functions used. To support the use of asyncio.run() like behavior for Python ≥ 3.5 while also supporting current Python versions, consider the boilerplates below.

asyncio.subprocess

A complete single-file example that works for Python ≥ 3.5 is date_coro.py. This supports AsyncIO subprocess.

def runner(fun, *args):
    """
    Generic asyncio.run() equivalent for Python >= 3.5
    """
    if os.name == 'nt' and (3, 7) <= sys.version_info < (3, 8):
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())  # type: ignore

    if sys.version_info >= (3, 7):
        result = asyncio.run(fun(*args))
    else:
        if os.name == 'nt':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.new_event_loop()
            asyncio.get_child_watcher().attach_loop(loop)
        result = loop.run_until_complete(fun(*args))
        loop.close()

    return result

asyncio.open_connection

For asyncio.open_connection, the ProactorEventLoop makes a lot of warnings with trapped RuntimeError. The workaround for this is to not use the ProactorEventLoop. This runner.py works for Python 3.5..3.8 at least.

def runner(fun, *args):
    """
    Generic asyncio.run() equivalent for Python >= 3.5 that
      does not use ProactorEventLoop on Windows
    """
    use_run = ((os.name == 'nt' and (3, 8) > sys.version_info >= (3, 7)) or
               (os.name != 'nt' and sys.version_info >= (3, 7)))

    if use_run:
        result = asyncio.run(fun(*args))
    else:  # 3.8, 3.6, 3.5
        loop = asyncio.SelectorEventLoop()
        result = loop.run_until_complete(fun(*args))
        loop.close()
    return result