How to detect task cancellation by Task Group

Given a taskgroup and number of running tasks, per taskgroup docs if any of the tasks raises an error, rest of the tasks in group will be cancelled.

If some of these tasks need to perform cleanup upon cancellation, then how would one go about detecting within the task it's being cancelled?

Was hoping some exception is raised in the task, but that's not the case:

script.py:

import asyncio

class TerminateTaskGroup(Exception):
    """Exception raised to terminate a task group."""

async def task_that_needs_to_cleanup_on_cancellation():
    try:
        await asyncio.sleep(10)
    except Exception:
        print('exception caught, performing cleanup...')


async def err_producing_task():
    await asyncio.sleep(1)
    raise TerminateTaskGroup()


async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task_that_needs_to_cleanup_on_cancellation())
            tg.create_task(err_producing_task())
    except* TerminateTaskGroup:
        print('main() termination handled')

asyncio.run(main())

Executing, we can see no exception is raised in task_that_needs_to_cleanup_on_cancellation():

$ python3 script.py 
main() termination handled

Solution

Casually, I might avoid a design where a task group is intentionally cancelled in this way

order and timing of Exception may be difficult to predict
potentially-awkward cleanup
new work created during cancellation

However, you can except asyncio.CancelledError or use a finally block

async def task_that_needs_to_cleanup_on_cancellation():
    try:
        await asyncio.sleep(10)  # placeholder for work
    except asyncio.CancelledError:
        print('task cancelled, performing cleanup...')
        raise  # end without running later logic

In order to avoid this cleanup-during-cancellation, you could pack potential cleanup work into a collection for the caller to handle in its Exception handler or later

Both designs feel a little ugly, but also are very specific to the individual cleanup case as I believe the Exception could occur during any await within them

async def task_1(task_id, cleanups):
    cleanups[task_id] = "args for later work"
    await asyncio.sleep(10)  # placeholder for work
    del cleanups[task_id]

async def task_2(name, cleanups):
    try:
        await asyncio.sleep(10)  # placeholder for work
    except asyncio.CancelledError:  # consider list instead
        cleanups[task_id] = "args for later work"
        raise

...
    cleanups = {}
    try:
         # task group logic
    except* TerminateTaskGroup:
        print('main() termination handled')

    if cleanups:
        # probably create and asyncio.gather()

Note that the docs also mention that CancelledError should be re-raised, but it may not matter if there is no further logic!
Additionally, when asyncio was added in Python 3.7, asyncio.CancelledError inherited from Exception - this was changed to BaseException in 3.8 (and is the case in all later versions)
https://docs.python.org/3/library/asyncio-exceptions.html