Learn Task Cancellation in Asyncio | Advanced Asyncio Patterns

Understanding how to cancel running tasks is crucial when working with asyncio. Sometimes, you may need to stop a task before it completes—perhaps because its result is no longer needed, it has taken too long, or your application is shutting down. In asyncio, you can cancel a task by calling its cancel() method. When a task is cancelled, it raises an asyncio.CancelledError inside the coroutine, which you can catch and handle if necessary. This allows you to clean up resources or perform any required finalization before the task fully stops. Handling cancellation properly ensures your asynchronous applications remain robust and responsive.


              12345678910111213141516171819202122
            
import asyncio

async def long_running_task():
    try:
        print("Task started.")
        await asyncio.sleep(5)
        print("Task finished.")
    except asyncio.CancelledError:
        print("Task was cancelled.")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(1)  # Let the task run for a bit
    print("Cancelling the task...")
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Caught task cancellation in main.")

await main()

In this example, you create a long-running task and then cancel it after one second. When the task is cancelled, it raises asyncio.CancelledError inside the coroutine, which is caught and handled. This pattern is essential for managing resources and ensuring your program responds properly to changes or user requests.

Everything was clear?

Thanks for your feedback!

Section 3. Chapter 1

Ask AI

Ask anything or try one of the suggested questions to begin our chat

Suggested prompts:

What happens if I don't handle asyncio.CancelledError in my coroutine?

Can you explain when I should use task cancellation in real applications?

Are there best practices for cleaning up resources after cancelling a task?

Swipe to show menu


              12345678910111213141516171819202122
            
import asyncio

async def long_running_task():
    try:
        print("Task started.")
        await asyncio.sleep(5)
        print("Task finished.")
    except asyncio.CancelledError:
        print("Task was cancelled.")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(1)  # Let the task run for a bit
    print("Cancelling the task...")
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Caught task cancellation in main.")

await main()

Everything was clear?

Thanks for your feedback!

Section 3. Chapter 1