python - 如何使用 Ctrl-C 优雅地终止 asyncio 脚本？

Question

我已经阅读了所有我能找到的关于如何使用 Ctrl-C 终止 asyncio 事件循环的脚本来优雅地处理脚本的帖子，而且我无法在不打印一个或多个的情况下让它们中的任何一个工作我这样做的回溯。答案几乎无处不在，但我无法将它们中的任何一个实现到这个小脚本中:

import asyncio
import datetime
import functools
import signal


async def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)


def stopper(signame, loop):
    print("Got %s, stopping..." % signame)
    loop.stop()


loop = asyncio.get_event_loop()
for signame in ('SIGINT', 'SIGTERM'):
    loop.add_signal_handler(getattr(signal, signame), functools.partial(stopper, signame, loop))

loop.run_until_complete(display_date(loop))
loop.close()

我想要的是脚本退出时不打印任何跟踪 Ctrl-C(或通过 kill 发送的 SIGTERM/SIGINT)。此代码打印 RuntimeError: Event loop stopped before Future completed. 在我根据之前的答案尝试过的许多其他形式中，我得到了大量其他类型的异常类和错误消息，没有想法如何修复它们。上面的代码现在是最小的，但我之前所做的一些尝试根本不是，而且没有一个是正确的。

如果您能够修改脚本以使其正常终止，那么将非常感谢解释为什么您的做法是正确的方式。

Answer 1

使用signal处理程序:

import asyncio
from signal import SIGINT, SIGTERM

async def main_coro():
    try:
        await awaitable()
    except asyncio.CancelledError:
        do_cleanup()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    main_task = asyncio.ensure_future(main_coro())
    for signal in [SIGINT, SIGTERM]:
        loop.add_signal_handler(signal, main_task.cancel)
    try:
        loop.run_until_complete(main_task)
    finally:
        loop.close()