异步运行或 run_until_complete

Question

我以非常基本的方式将asyncio 用于应用程序。检查互联网上的大多数教程（甚至是官方文档），我发现他们使用get_event_loop() 和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()

但是在Python 3.7 docs，我们可以阅读：

应用程序开发人员通常应该使用高级异步函数，例如asyncio.run()，并且很少需要引用循环对象或调用其方法。本部分主要面向需要更好地控制事件循环行为的低级代码、库和框架的作者。

我发现它更加简洁易用，但它仅适用于 Python 3.7+。所以在这里我必须做出选择，是使用 Python 3.7+ 和run()，还是让它与 Python 3.6 兼容并使用事件循环。你将如何管理这个？有没有一种简单的方法可以使其向后兼容 Python 3.6？在 Python 3.7 成为通用版本之前，我是否应该先检查 Python 版本并基于此使用一种或另一种方式？

Answer 1

有没有一种简单的方法使 [使用asyncio.run 的代码] 向后兼容 Python 3.6？

您可以实现asyncio.run 的简单替换，并在较旧的 Python 版本上调用它：

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)

与仅使用 loop.run_until_complete() 相比，这种方法的优势在于，即使在较旧的 Python 版本上，您也可以在接近新 asyncio.run 的语义下执行代码。 （例如，您将始终在新创建的事件循环上运行。）放弃对 3.7 之前的 Python 的支持就像删除 run shim 并直接调用 asyncio.run 一样简单。

Answer 2

可以通过从 asyncio.runners.py 复制代码来复制 asyncio.run。下面是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,
            })