asyncio.TaskGroup中KeyboardInterrupt无法捕获的问题及正确检测方法

根据Python官方Task Groups文档说明：

有两种基础异常会被特殊处理：如果任何任务抛出KeyboardInterrupt或SystemExit，任务组仍会取消剩余任务并等待，但之后会重新抛出最初的KeyboardInterrupt或SystemExit，而非ExceptionGroup或BaseExceptionGroup。

我原以为运行以下代码并按下Ctrl-C时会打印Stopped：

import asyncio

async def task():
    await asyncio.sleep(10)

async def run() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(task())
            t2 = tg.create_task(task())
        print("Done")
    except KeyboardInterrupt:
        print("Stopped")

asyncio.run(run())

但实际运行时异常并未被捕获，报错信息如下：

^CTraceback (most recent call last):
  File "<python>/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<python>/asyncio/base_events.py", line 685, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "<module>/__init__.py", line 8, in run
    async with asyncio.TaskGroup() as tg:
  File "<python>/asyncio/taskgroups.py", line 134, in __aexit__
    raise propagate_cancellation_error
  File "<python>/asyncio/taskgroups.py", line 110, in __aexit__
    await self._on_completed_fut
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<frozen runpy>", line 189, in _run_module_as_main
  File "<frozen runpy>", line 148, in _get_module_details
  File "<frozen runpy>", line 112, in _get_module_details
  File "<module>/__init__.py", line 15, in <module>
    asyncio.run(run())
  File "<python>/asyncio/runners.py", line 194, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "<python>/asyncio/runners.py", line 123, in run
    raise KeyboardInterrupt()
KeyboardInterrupt

请问我忽略了什么？正确检测KeyboardInterrupt的方式是什么？

你忽略的核心点 #

asyncio.run()有个设计特性：它会主动重新抛出KeyboardInterrupt和SystemExit，而且是在你的run()函数执行完毕之后。也就是说，你的内部try/except其实已经捕获到了任务组抛出的KeyboardInterrupt，但asyncio.run()会把这个异常再次抛到主线程，导致你看到最终的报错栈。

从报错信息里也能看出来：最后是asyncio/runners.py里的run()方法主动抛出了KeyboardInterrupt()，这不是任务组的问题，而是asyncio.run()的默认行为。

正确的处理方式 #

要成功捕获这个异常，你需要把try/except块放在asyncio.run()的外面，而不是内部的异步函数里：

import asyncio

async def task():
    await asyncio.sleep(10)

async def run() -> None:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task())
        t2 = tg.create_task(task())
    print("Done")

if __name__ == "__main__":
    try:
        asyncio.run(run())
    except KeyboardInterrupt:
        print("Stopped")

修改后按下Ctrl-C，就能正常打印Stopped了。

如果一定要在异步函数内部处理，你需要同时捕获CancelledError和KeyboardInterrupt，但这不是最优方案——因为任务组在响应取消时会先抛出CancelledError，而且asyncio.run()最终还是会重新抛出KeyboardInterrupt。所以最直接可靠的方式还是在asyncio.run()外部捕获异常。

内容的提问来源于stack exchange，提问作者Amadan