The docs on Task Groups say:
Two base exceptions are treated specially: If any task fails with
KeyboardInterrupt
orSystemExit
, the task group still cancels the remaining tasks and waits for them, but then the initialKeyboardInterrupt
orSystemExit
is re-raised instead ofExceptionGroup
orBaseExceptionGroup
.
This makes me believe, given the following code:
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())
running and hitting Ctrl-C should result in printing Stopped
; but in fact, the exception is not caught:
^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
What am I missing? What is the correct way of detecting KeyboardInterrupt
?