Managing and Cancelling Asyncio Tasks and Sub‑coroutines in Python

Introduction In Python asynchronous programming, the asyncio library offers various ways to manage and cancel tasks, as well as to call and control sub‑coroutines.

1. Cancel Task The basic method to cancel a task is to call Task.cancel() . The following example shows how to create a task, wait briefly, then cancel it and handle the CancelledError :

import asyncio

async def my_coroutine():
    print("Coroutine started")
    await asyncio.sleep(2)
    print("Coroutine finished")

async def main():
    task = asyncio.create_task(my_coroutine())
    await asyncio.sleep(1)
    print("Cancelling the task...")
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled")

asyncio.run(main())

2. Sub‑coroutine Call Sub‑coroutines can be invoked directly (without awaiting) or with await to wait for completion. Two examples illustrate both approaches.

Direct call (no await):

import asyncio

async def sub_coroutine():
    print("Sub-coroutine started")
    await asyncio.sleep(1)
    print("Sub-coroutine finished")

async def main_coroutine():
    print("Main-coroutine started")
    sub_coroutine()  # Direct call, does not wait
    await asyncio.sleep(2)
    print("Main-coroutine finished")

asyncio.run(main_coroutine())

Using await (waits for sub‑coroutine):

import asyncio

async def sub_coroutine():
    print("Sub-coroutine started")
    await asyncio.sleep(1)
    print("Sub-coroutine finished")

async def main_coroutine():
    print("Main-coroutine started")
    await sub_coroutine()  # Await ensures completion
    print("Main-coroutine finished")

asyncio.run(main_coroutine())

3. Cancel Sub‑coroutine A sub‑coroutine can be cancelled from the main coroutine using Task.cancel() . The example demonstrates creating a task for the sub‑coroutine, cancelling it after a short delay, and handling the cancellation.

import asyncio

async def sub_coroutine(task):
    print("Sub-coroutine started")
    try:
        await asyncio.sleep(3)
        print("Sub-coroutine finished")
    except asyncio.CancelledError:
        print("Sub-coroutine cancelled")

async def main_coroutine():
    print("Main-coroutine started")
    sub_task = asyncio.create_task(sub_coroutine(asyncio.current_task()))
    await asyncio.sleep(1)
    print("Cancelling sub-coroutine...")
    sub_task.cancel()
    try:
        await sub_task
    except asyncio.CancelledError:
        print("Sub-coroutine cancelled")
    print("Main-coroutine finished")

asyncio.run(main_coroutine())

4. Cancel Multiple Tasks When several sub‑coroutines need to be cancelled, iterate over the list of tasks and call cancel() on each, then gather results with return_exceptions=True :

import asyncio

async def sub_coroutine(task):
    print(f"Sub-coroutine {task.get_name()} started")
    try:
        await asyncio.sleep(3)
        print(f"Sub-coroutine {task.get_name()} finished")
    except asyncio.CancelledError:
        print(f"Sub-coroutine {task.get_name()} cancelled")

async def main_coroutine():
    print("Main-coroutine started")
    tasks = [
        asyncio.create_task(sub_coroutine(asyncio.current_task()), name="Task-1"),
        asyncio.create_task(sub_coroutine(asyncio.current_task()), name="Task-2"),
        asyncio.create_task(sub_coroutine(asyncio.current_task()), name="Task-3")
    ]
    await asyncio.sleep(1)
    print("Cancelling all tasks...")
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    print("Main-coroutine finished")

asyncio.run(main_coroutine())

5. Sub‑coroutine Error Handling To propagate and handle exceptions raised inside a sub‑coroutine, catch the exception, log it, and re‑raise so the main coroutine can process it:

import asyncio

async def sub_coroutine(task):
    print(f"Sub-coroutine {task.get_name()} started")
    try:
        await asyncio.sleep(1)
        raise ValueError("An error occurred")
    except Exception as e:
        print(f"Error in sub-coroutine {task.get_name()}: {e}")
        raise

async def main_coroutine():
    print("Main-coroutine started")
    tasks = [
        asyncio.create_task(sub_coroutine(asyncio.current_task()), name="Task-1"),
        asyncio.create_task(sub_coroutine(asyncio.current_task()), name="Task-2"),
        asyncio.create_task(sub_coroutine(asyncio.current_task()), name="Task-3")
    ]
    await asyncio.gather(*tasks, return_exceptions=True)
    print("Main-coroutine finished")

asyncio.run(main_coroutine())

6. Summary The article covered the principles and practical code for cancelling asyncio tasks, invoking sub‑coroutines with and without await , cancelling sub‑coroutines, cancelling multiple tasks simultaneously, and handling exceptions raised by sub‑coroutines.