Timeouts in asynchronous programs are easy to scatter and surprisingly hard to compose. A service call gets five seconds, a database query gets five more, and a retry gets another five. Each individual limit looks reasonable, yet the whole request can run far beyond the caller’s budget.

Python 3.11 added asyncio.timeout(), an asynchronous context manager that makes a different model practical: put a time budget around a block of work, not just around one awaitable.

The useful idea is to treat a deadline as part of an operation’s lifetime. Nested functions can still have smaller local limits, but they should not accidentally reset the outer budget.

Put the timeout around the operation

A basic timeout is straightforward:

import asyncio


async def load_dashboard(user_id: str):
    try:
        async with asyncio.timeout(2.0):
            profile = await load_profile(user_id)
            alerts = await load_alerts(user_id)
            return profile, alerts
    except TimeoutError:
        return None

The two awaits share one two-second budget. If load_profile() consumes 1.6 seconds, load_alerts() does not receive a fresh two seconds.

That is different from wrapping each operation independently:

profile = await asyncio.wait_for(load_profile(user_id), 2.0)
alerts = await asyncio.wait_for(load_alerts(user_id), 2.0)

Those calls can consume roughly four seconds before accounting for cancellation and cleanup. Per-call limits can still be useful, but they are not a substitute for an end-to-end budget.

Catch TimeoutError outside the context

The cancellation mechanics matter. When the deadline expires, the timeout context manager cancels the current task. It handles the resulting asyncio.CancelledError internally and converts it to TimeoutError as the context exits.

That means the normal catch point is outside the async with block:

async def fetch_report():
    try:
        async with asyncio.timeout(1.5):
            return await build_report()
    except TimeoutError:
        return {'status': 'timed-out'}

Do not write code that assumes a TimeoutError will appear at an arbitrary await inside the block. Code running there observes cancellation first.

This distinction also helps keep cancellation handling honest. A coroutine should generally use try/finally for cleanup and let CancelledError propagate unless it has a specific reason to do otherwise:

async def use_connection(pool):
    conn = await pool.acquire()
    try:
        return await conn.read_result()
    finally:
        await pool.release(conn)

If lower-level code swallows cancellation, the timeout boundary cannot provide the lifetime semantics its caller expects.

Prefer absolute deadlines when a budget crosses layers

Relative durations are convenient at an API boundary, but repeatedly passing timeout=2.0 down a call stack can reset the clock. An absolute deadline avoids that mistake.

asyncio.timeout_at() accepts a deadline measured with the event loop’s monotonic clock:

import asyncio


async def handle_request(request):
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 2.0

    try:
        async with asyncio.timeout_at(deadline):
            return await assemble_response(request, deadline)
    except TimeoutError:
        return {'status': 504}

A lower layer can derive the remaining budget without inventing a new one:

async def assemble_response(request, deadline: float):
    loop = asyncio.get_running_loop()
    remaining = max(0.0, deadline - loop.time())

    async with asyncio.timeout(min(remaining, 0.5)):
        metadata = await load_metadata(request)

    return await render_response(metadata)

The half-second limit is a local cap, while the outer absolute deadline remains authoritative.

Use loop.time() for these calculations rather than wall-clock timestamps. Deadline scheduling needs a monotonic clock so ordinary system-clock adjustments do not change the meaning of elapsed time.

Reschedule when the deadline is learned late

Sometimes an operation begins before its actual budget is known. asyncio.timeout(None) starts with no deadline, and the returned Timeout object can later be rescheduled:

import asyncio


async def process_message(message):
    loop = asyncio.get_running_loop()

    try:
        async with asyncio.timeout(None) as timeout:
            policy = await load_policy(message.account_id)
            timeout.reschedule(loop.time() + policy.max_seconds)
            await apply_message(message)
    except TimeoutError:
        await record_timeout(message.id)

This is more precise than leaving a guessed timeout in place. The same object exposes when() to inspect its current deadline and expired() to determine whether the context actually exceeded it.

Be careful about what work happens before rescheduling. In the example, load_policy() is intentionally outside the eventual policy budget because the deadline is unknown until that call completes. If policy loading itself needs a bound, give that phase a separate explicit limit.

Nested timeout scopes are valid

Timeout contexts can be nested. This is useful when an outer request has a total budget but one phase deserves a stricter cap:

async def serve():
    async with asyncio.timeout(3.0):
        async with asyncio.timeout(0.4):
            cache_value = await query_cache()

        return await compute_response(cache_value)

The inner scope cannot extend the outer deadline. Whichever deadline is reached first controls the relevant scope.

In real code, decide what each timeout means. An inner cache timeout might mean “fall back to the database,” while the outer request timeout might mean “stop all request work.” Those are different policies and deserve different handling boundaries.

A timeout is not proof that side effects did not happen

Cancellation is a control-flow signal, not a transaction rollback.

Suppose an async client sends a payment request and then times out while waiting for the response:

async with asyncio.timeout(1.0):
    receipt = await payment_client.charge(order)

A timeout does not establish whether the remote system received, committed, or rejected the charge. Retrying blindly can duplicate the side effect.

For operations with externally visible effects, combine timeouts with the protocol’s correctness mechanisms: idempotency keys, request identifiers, deduplication, transaction boundaries, or a follow-up status query. The timeout only says that this task did not complete the awaited operation within its budget.

Cancellation cleanup can affect observed latency

A deadline is not a guarantee that the surrounding function returns at the exact deadline. Cancellation has to be delivered and code may need to unwind cleanup.

For example:

async def worker(resource):
    try:
        await resource.run()
    finally:
        await resource.close()

The cleanup is important, but it can take time. If shutdown itself can block indefinitely, it needs its own design rather than an assumption that the original timeout makes every cleanup operation bounded.

This is one reason latency objectives and timeout values should not be treated as identical concepts. A timeout is a cancellation policy. End-to-end latency also includes scheduling, cancellation response, cleanup, serialization, and any fallback work after the timeout is caught.

Do not accidentally convert caller cancellation into a local timeout

An async task can be cancelled for reasons unrelated to its own timeout: server shutdown, client disconnect, parent-task cancellation, or structured-concurrency failure.

Code should preserve that distinction. Catch TimeoutError at the timeout boundary when you want to handle expiry of that scope. Avoid broad exception handling that treats every cancellation path as “the dependency timed out.”

In particular, asyncio.CancelledError is part of cooperative task cancellation. Cleanup code can observe it, but suppressing it without a deliberate ownership decision can make task groups, shutdown, and timeout scopes behave incorrectly.

Test behavior, not tiny wall-clock margins

Timing tests become flaky when they assert that something completed within an extremely narrow real-time interval. Prefer observable state and generous timing separation.

For example:

import asyncio
import pytest


@pytest.mark.asyncio
async def test_timeout_cancels_the_operation():
    cleaned_up = asyncio.Event()

    async def slow_operation():
        try:
            await asyncio.sleep(60)
        finally:
            cleaned_up.set()

    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.05):
            await slow_operation()

    assert cleaned_up.is_set()

This test checks the contract that matters: the timeout expires, the operation is cancelled, and its cleanup runs. It does not claim that a busy CI worker will return at an exact millisecond.

Also test the non-timeout path, nested policies, and any retry or fallback behavior. For side-effecting integrations, test the idempotency mechanism independently of timeout delivery.

A practical design rule

I use a few rules when adding time limits to async Python code:

  • Put an outer timeout around the complete unit of work whose latency you actually want to bound.
  • Use absolute monotonic deadlines when a budget crosses multiple layers.
  • Add smaller nested limits only when they represent a real local policy.
  • Catch TimeoutError outside the timeout context that owns it.
  • Let cancellation propagate through lower-level code after necessary cleanup.
  • Never interpret a timeout as proof that a remote side effect did not occur.
  • Test cancellation and cleanup semantics instead of relying on exact elapsed-time assertions.

asyncio.timeout() is small API surface, but it encourages a useful shift in thinking. The question stops being “how long may this one await take?” and becomes “what lifetime does this operation own?” That is usually the better boundary for reliable async systems.

Python Asyncio Concurrency Timeouts Standard Library