Cancelling Tasks

Summary: in this tutorial, you’ll learn how to cancel a long-running asynchronous operation that may take forever to complete.

The following statement uses the await statement to wait for a task to be complete:

task = asyncio.create_task(coroutine()) result = await taskCode language: Python (python)

However, if the coroutine() took forever, you would be stuck waiting for the await statement to finish without obtaining any result. Additionally, you would have no way to stop it if you wanted to.

To resolve this, you can cancel the task using the cancel() method of the Task object. If you cancel a task, it’ll raise the CancelledError exception when you await it. For example:

import asyncio from asyncio import CancelledError async def call_api(message, result=1000, delay=3): print(message) await asyncio.sleep(delay) return result async def main(): task = asyncio.create_task( call_api('Calling API...', result=2000, delay=5) ) if not task.done(): print('Cancelling the task...') task.cancel() try: await task except CancelledError: print('Task has been cancelled.') asyncio.run(main())Code language: Python (python)

Output:

Cancelling the task... Task has been cancelledCode language: Python (python)

How it works.

First, the call_api() coroutine prints a message, delays 3 seconds, and returns the result.

Second, create a new task using the create_task() function and pass the call_api() coroutine. The task will take 5 seconds to complete:

task = asyncio.create_task( call_api('Calling API...', result=2000, delay=5) )Code language: Python (python)

Third, check if the task is not done by calling the done() method and cancel the task using the cancel() method:

if not task.done(): print('Cancelling the task...') task.cancel()Code language: Python (python)

Finally, wait for the task to be completed using the await keyword. Since the task has been canceled, the CancelledError exception is raised:

try: await task except CancelledError: print('Task has been cancelled.')Code language: Python (python)

If you want to check every second if a task has been completed and cancel it if an amount of time has passed, you can use a while loop:

import asyncio from asyncio import CancelledError async def call_api(message, result=1000, delay=3): print(message) await asyncio.sleep(delay) return result async def main(): task = asyncio.create_task( call_api('Calling API...', result=2000, delay=5) ) time_elapsed = 0 while not task.done(): time_elapsed += 1 await asyncio.sleep(1) print('Task has not completed, checking again in a second') if time_elapsed == 3: print('Cancelling the task...') task.cancel() break try: await task except CancelledError: print('Task has been cancelled.') asyncio.run(main())Code language: Python (python)

In this example, the while loop checks if the task has been completed every second and cancels the task once the elapsed time reaches 3 seconds.