Python: Handling asyncio.exceptions.CancelledError gracefully

What is the Point?

In modern Python (3.8 and newer), asyncio.exceptions.CancelledError is an exception that is raised when an asynchronous operation has been cancelled. It is a subclass of BaseException. Some possible scenarios that can cause this exception are:

• Calling the cancel() method on a Task or Future object.
• Using the asyncio.wait_for() function with a timeout and the timeout expires.
• Using the asyncio.sleep() function and the sleep is cancelled by another coroutine or task.
• Using the asyncio.Queue.get() function and the queue is empty.

If you don’t handle the asyncio.exceptions.CancelledError properly, it may cause unwanted errors or warnings in your program, or hide the cancellation from other parts of your code. The following examples will give you some general guidelines to deal with it gracefully (with try/except blocks).

Examples

Cancelling a task and catching the exception (basic)

If you are using asyncio.run() to run a main coroutine, you should catch the exception in the main coroutine and perform any necessary cleanup actions.

This example shows how to create a task from a coroutine using asyncio.create_task(), and how to cancel it using task.cancel():

# SlingAcademy.com
# This code uses Python 3.11.4

import asyncio


async def say_hello():
    try:
        print("Hello buddy.")
        await asyncio.sleep(1)
        print("Welcome to Sling Academy.")
    except asyncio.CancelledError:
        print("Cancelled")
        # re-raise the exception to propagate it to the caller
        raise


async def main():
    # create a task from the say_hello coroutine
    task = asyncio.create_task(say_hello())
    # wait for half a second
    await asyncio.sleep(0.5)
    # cancel the task before it finishes
    task.cancel()
    # wait for the task to finish
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled before it finished.")
        print(
            "The second print statement (Welcome to Sling Academy!) was never executed."
        )


asyncio.run(main())

The say_hello coroutine catches the CancelledError exception and prints a message before re-raising it. The main coroutine also catches the exception and prints another message.

Output:

Hello buddy.
Cancelled
Task was cancelled before it finished.
The second print statement (Welcome to Sling Academy!) was never executed.

Cancelling multiple tasks with a loop (intermediate)

If you are using asyncio.gather() to run multiple coroutines or tasks concurrently, you can use the return_exceptions parameter to treat exceptions as successful results and aggregate them in the result list. This way, you can inspect the results and handle any cancellation exceptions accordingly.

This example shows how to create multiple tasks from the same coroutine with different arguments, and how to cancel them all using a loop. In the main coroutine, we’ll use asyncio.gather() to wait for all tasks and ignores the exceptions by passing return_exceptions=True:

# SlingAcademy.com
# This code uses Python 3.11.4

import asyncio
import random

# create a worker coroutine
async def worker(name):
    try:
        print(f"{name} started")
        # simulate some work
        await asyncio.sleep(random.randint(1, 5))
        print(f"{name} finished")
    except asyncio.CancelledError:
        print(f"{name} cancelled")
        raise

# main coroutine
async def main():
    # create four tasks from the worker coroutine
    tasks = [asyncio.create_task(worker(f"worker-{i}")) for i in range(4)]
    # wait for two seconds
    await asyncio.sleep(2)
    # cancel all tasks
    for task in tasks:
        task.cancel()
    # use asyncio.gather() to wait for all tasks and ignore the exceptions
    await asyncio.gather(*tasks, return_exceptions=True)


asyncio.run(main())

Output (yours might be different from mine due to the randomness):

worker-0 started
worker-1 started
worker-2 started
worker-3 started
worker-3 finished
worker-0 cancelled
worker-1 cancelled
worker-2 cancelled

Cancelling a blocking operation using another coroutine (advanced)

If you are creating a custom coroutine or task that may be cancelled by another part of your code, you should catch the exception in the coroutine or task and perform any necessary cleanup actions. You should also re-raise the exception to propagate the cancellation to other coroutines or tasks that may depend on it.

This example demonstrates how to cancel a blocking operation, such as asyncio.sleep(), using a different coroutine. The code is a little bit longer than the preceding examples:

# SlingAcademy.com
# This code uses Python 3.11.4

import asyncio

# create a coroutine named sleeper
async def sleeper():
    try:
        print("Going to sleep")
        # sleep for 10 seconds or until cancelled
        await asyncio.sleep(10)
        print("Woke up")
    except asyncio.CancelledError:
        print("Interrupted")
        raise

# create a coroutine named canceller
async def canceller():
    # wait for 3 seconds
    await asyncio.sleep(3)
    # get the current event loop
    loop = asyncio.get_running_loop()
    # get all the tasks in the loop
    tasks = asyncio.all_tasks(loop)
    # find the sleeper task (assuming there is only one)
    for task in tasks:
        if task.get_coro().__name__ == "sleeper":
            # cancel the sleeper task
            task.cancel()
            break

# main coroutine
async def main():
    # create two tasks: one for sleeper and one for canceller
    tasks = [asyncio.create_task(sleeper()), asyncio.create_task(canceller())]
    # use asyncio.wait() to wait for both tasks to finish or be cancelled
    done, pending = await asyncio.wait(tasks)
    # print the result or exception of each task
    for task in done:
        try:
            result = task.result()
            print(f"Task result: {result}")
        except asyncio.CancelledError:
            print("Task was cancelled")


asyncio.run(main())

The sleeper coroutine tries to sleep for 10 seconds, but is interrupted by the canceller coroutine after 3 seconds. The canceller coroutine finds the sleeper task in the event loop and cancels it. The main coroutine uses asyncio.wait() to wait for both tasks to finish or be cancelled, and prints the result or exception of each task.

Output:

Going to sleep
Interrupted
Task result: None
Task was cancelled

That’s it. This tutorial ends here. Happy coding with Python & continue building your big projects!