How to Use asyncio.timeout in Modern Python

This article covers how to implement and manage timeouts in modern asynchronous Python using the asyncio.timeout() context manager introduced in Python 3.11. You will learn the mechanics of how this context manager cancels long-running tasks, how to handle the resulting TimeoutError, how to dynamically adjust deadlines during execution, and why it provides a cleaner, more robust alternative to older approaches like asyncio.wait_for().

The Basics of asyncio.timeout()

In Python 3.11 and later, asyncio.timeout() serves as the standard way to apply a deadline to one or more asynchronous operations. It is used as an asynchronous context manager that accepts a single argument: the timeout duration in seconds (or None to disable the timeout).

Here is a basic implementation:

import asyncio

async def fetch_data():
    await asyncio.sleep(3)
    return "Data fetched"

async def main():
    try:
        async with asyncio.timeout(2):
            result = await fetch_data()
            print(result)
    except TimeoutError:
        print("The operation timed out.")

asyncio.run(main())

In this example, fetch_data() attempts to sleep for 3 seconds, but the context manager enforces a 2-second limit. When the limit is reached, the context manager cancels the running task within its scope and raises a standard TimeoutError.

How the Cancellation Flow Works

Under the hood, asyncio.timeout() schedules a cancellation callback on the event loop when entering the context block.

Entering the context (async with): The context manager calculates an absolute deadline using the current event loop time plus the specified delay (loop.time() + delay).
Normal completion: If the code inside the block finishes before the deadline, the context manager cancels the scheduled timeout callback and exits cleanly.
Reaching the deadline: If the timer expires while code inside the block is awaiting, the context manager calls task.cancel() on the enclosing task.
Exiting the context: As the task unwinds due to cancellation, the context manager catches the internal asyncio.CancelledError, checks if it was caused by this specific timeout, and converts it into a TimeoutError.

Because the conversion from CancelledError to TimeoutError occurs when leaving the context manager, you must place your try...except TimeoutError block outside the async with asyncio.timeout() block.

Dynamic Deadline Rescheduling

The context manager returns a Timeout object that allows you to inspect and modify the deadline while the block is running.

import asyncio

async def main():
    try:
        async with asyncio.timeout(5) as cm:
            # Check the absolute deadline
            print(f"Deadline scheduled for: {cm.when()}")
            
            await asyncio.sleep(1)
            
            # Extend the deadline by shifting it forward
            new_deadline = asyncio.get_running_loop().time() + 10
            cm.reschedule(new_deadline)
            
            # Check if expired
            print(f"Has the timeout expired? {cm.expired()}")
    except TimeoutError:
        print("Timed out.")

asyncio.run(main())

Useful methods and attributes of the Timeout object include:

cm.when(): Returns the absolute monotonic clock deadline, or None if no deadline is set.
cm.reschedule(when): Updates the deadline. Passing an absolute timestamp (from loop.time()) shifts the deadline, while passing None removes the timeout entirely.
cm.expired(): Returns True if the timeout triggered a cancellation.

Absolute Timeouts with asyncio.timeout_at()

If you already have a calculated target timestamp from loop.time(), use asyncio.timeout_at() instead of calculating relative seconds:

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 2.5

    try:
        async with asyncio.timeout_at(deadline):
            await asyncio.sleep(5)
    except TimeoutError:
        print("Reached absolute deadline.")

asyncio.run(main())

Advantages Over asyncio.wait_for()

Before Python 3.11, the primary tool for timing out operations was asyncio.wait_for(aw, timeout). The asyncio.timeout() context manager improves on this pattern in several ways:

Scoped Blocks: asyncio.wait_for requires wrapping a single awaitable or task. asyncio.timeout() can enclose a block containing multiple await expressions, sequential requests, or custom logic without combining them into a single coroutine.
Cleaner Cancellation Semantics: asyncio.wait_for creates an intermediate task to manage the timeout, which often led to swallowed cancellation exceptions or subtle bugs when combined with task cancellation trees. asyncio.timeout() works directly on the current task.
Consistent Exception Types: asyncio.timeout() raises the built-in TimeoutError directly, aligning with Python's unified exception hierarchy.