Timeouts and Cancellation

Why timeouts matter

Network calls can stall.

Always use timeouts to avoid:

• stuck tasks
• resource leaks

asyncio.wait_for

import asyncio
 
 
async def slow():
    await asyncio.sleep(5)
    return "ok"
 
 
async def main():
    try:
        result = await asyncio.wait_for(slow(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("timed out")
 
 
asyncio.run(main())

Cancelling a task

import asyncio
 
 
async def worker():
    try:
        while True:
            print("working")
            await asyncio.sleep(0.2)
    except asyncio.CancelledError:
        print("cancelled")
        raise
 
 
async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.7)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("task cancelled confirmed")
 
 
asyncio.run(main())

Best practices

• Consider timeouts for all external calls.
• Always handle cancellation cleanly.

Why you need a timeout

Waiting is cheap in asyncio, so it is easy to wait forever. A server that never answers will hold your task, its memory, and possibly a connection open indefinitely. A timeout turns "wait forever" into "give up after N seconds and do something sensible".

How the two examples work

wait_for(slow(), timeout=1) runs slow() but stops waiting after one second. When time runs out, asyncio cancels slow() and raises asyncio.TimeoutError to you (in Python 3.11+ this is the same as the built-in TimeoutError). That is why the example prints timed out after about one second, not five.

Cancellation works by raising asyncio.CancelledError inside the task at its current await. In cancel_task.py, the worker prints "working" about four times, is cancelled at 0.7 seconds, prints "cancelled" from its except block, and re-raises the error. Then await task in main raises CancelledError too, confirming the task really ended.

The modern style

From Python 3.11 you can also write a timeout as a block, which is handy when several awaits share one deadline:

import asyncio
 
 
async def main() -> None:
    try:
        async with asyncio.timeout(0.5):
            await asyncio.sleep(0.1)
            print("step 1 done")
            await asyncio.sleep(1)  # this one exceeds the deadline
            print("step 2 done")
    except TimeoutError:
        print("gave up")
 
 
asyncio.run(main())

Only "step 1 done" and "gave up" are printed.

Common mistakes

• Swallowing CancelledError. Catch it only to clean up (close a file, release a resource) and then raise it again. If you swallow it, the task cannot be stopped.
• Expecting a timeout to interrupt blocking code. Cancellation only happens at an await. A time.sleep(5) cannot be interrupted. Move blocking work to asyncio.to_thread.
• No timeout on external calls. Every network call should have one.
• Assuming the caller giving up stops the work. With asyncio.shield, the caller gets a timeout but the inner task keeps running.

When to use it

Wrap any call to something outside your program: HTTP requests, database queries, or waiting for a message. Use task.cancel() for background tasks that should stop when the program shuts down, and always await them afterwards so their cleanup code runs.

🧪 Try It Yourself

Exercise 1 – asyncio.wait_for Timeout

Press Run to see what your code prints.

Exercise 2 – Cancel a Task

Press Run to see what your code prints.

Exercise 3 – Check Cancellation

Press Run to see what your code prints.