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 aTask
orFuture
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!