Asynchronous operations need limits. A network call may never respond, a queue may remain empty, and a slow dependency may consume the entire request budget. asyncio.timeout() creates a deadline scope: if the block does not finish in time, the current task is cancelled and the context converts that cancellation into TimeoutError.
This guide covers relative and absolute deadlines, rescheduling, expiration checks, TaskGroup integration, comparison with asyncio.wait_for(), cleanup, shielding, retries, and common cancellation mistakes.
Your first timeout
import asyncio
async def slow_operation() -> str:
await asyncio.sleep(5)
return "ok"
async def main():
try:
async with asyncio.timeout(1.0):
result = await slow_operation()
print(result)
except TimeoutError:
print("deadline exceeded")
asyncio.run(main())After one second, the context cancels the current task. When the block exits, the internal cancellation is transformed into TimeoutError, which should be caught outside the context.
Why catch outside the block?
The conversion from CancelledError to TimeoutError happens when the context manager exits.
try:
async with asyncio.timeout(1.0):
await operation()
except TimeoutError:
...This also makes it clear which scope owns the deadline.
Timeout cannot interrupt blocking code
Asyncio cancellation is cooperative. A timeout can act only when the coroutine yields control to the event loop. CPU-bound code or a direct blocking call prevents timely cancellation.
async def incorrect():
time.sleep(10) # blocks the event loopUse asyncio.to_thread(), an executor, or a process for blocking work.
async with asyncio.timeout(2.0):
result = await asyncio.to_thread(blocking_function)Cancelling the await stops waiting for the thread, but it does not necessarily stop the underlying function. Give blocking libraries their own native timeout when possible.
Relative timeout and absolute deadline
asyncio.timeout(seconds) defines a relative limit. asyncio.timeout_at(when) receives an absolute value from the event loop’s monotonic clock.
loop = asyncio.get_running_loop()
deadline = loop.time() + 3.0
async with asyncio.timeout_at(deadline):
await step_a()
await step_b()Absolute deadlines are useful when several layers must share one budget rather than restarting the timer.
Propagating a time budget
If a request has five seconds total, each nested function should not receive a new five-second allowance.
async def query_a(deadline: float):
async with asyncio.timeout_at(deadline):
return await call_a()
async def flow():
loop = asyncio.get_running_loop()
deadline = loop.time() + 5.0
a = await query_a(deadline)
b = await query_b(deadline)
return a, bTime spent in the first step automatically reduces what remains for the second.
Rescheduling a timeout
The timeout object exposes reschedule(). This is useful when the deadline becomes known only after reading metadata or negotiating with another service.
async def process():
loop = asyncio.get_running_loop()
async with asyncio.timeout(None) as control:
limit = await get_limit()
control.reschedule(loop.time() + limit)
return await run_work()None creates a context without an initial deadline. Rescheduled times use the loop’s monotonic clock.
Checking expiration
After the context, expired() reports whether the deadline was reached.
control = None
try:
async with asyncio.timeout(1.0) as control:
await work()
except TimeoutError:
pass
if control is not None and control.expired():
print("the deadline expired")Catching TimeoutError is normally enough; expired() is useful for metrics and diagnostics.
Nested timeouts
Timeout contexts may be safely nested. The shorter deadline fires first.
async with asyncio.timeout(10.0):
await fast_step()
try:
async with asyncio.timeout(1.0):
await optional_call()
except TimeoutError:
use_fallback()
await final_step()The inner timeout can be handled locally while the outer timeout protects the total request budget.
Combining timeout with TaskGroup
A timeout around TaskGroup applies to the whole concurrent unit.
async def dashboard():
try:
async with asyncio.timeout(3.0):
async with asyncio.TaskGroup() as group:
user = group.create_task(get_user())
orders = group.create_task(get_orders())
except TimeoutError:
return {"error": "services were slow"}
return {
"user": user.result(),
"orders": orders.result(),
}When the deadline expires, the containing task is cancelled; TaskGroup cancels children and waits for cleanup.
Per-task limits inside a group
Put a timeout inside each child when operations have separate limits.
async def query_with_limit(name: str, seconds: float):
async with asyncio.timeout(seconds):
return await query(name)If TimeoutError escapes, TaskGroup treats it as a failure and cancels siblings. If timeout is an expected result, catch it inside the child and return an explicit status.
Comparison with asyncio.wait_for
asyncio.wait_for(awaitable, timeout) wraps one awaitable. asyncio.timeout() wraps an entire block with multiple awaits and intermediate logic.
result = await asyncio.wait_for(operation(), timeout=2.0)
async with asyncio.timeout(2.0):
header = await read_header()
body = await read_body(header)
await validate(body)The context manager is often clearer for composed flows. wait_for() remains convenient for one operation.
Cleanup and finally
Cancelled coroutines must release resources.
async def use_connection():
connection = await open_connection()
try:
return await connection.receive()
finally:
await connection.close()The timeout waits while cancellation propagates and cleanup runs. A slow cleanup can make observed duration exceed the nominal deadline. Keep shutdown bounded and predictable.
Do not swallow CancelledError
Inside the operation, timeout is implemented with cancellation. Catching CancelledError and returning normally may prevent the timeout context from recognizing expiration.
async def bad():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
return "ignored"Perform cleanup and re-raise.
except asyncio.CancelledError:
await cleanup()
raiseShielding critical work
asyncio.shield() can protect an inner task from cancellation, but ownership must remain explicit.
task = asyncio.create_task(write_audit())
try:
async with asyncio.timeout(1.0):
await asyncio.shield(task)
except TimeoutError:
...
await taskThe timeout stops waiting, but the protected task continues. Keep a strong reference and define who will await it. Shield can otherwise recreate detached tasks.
Timeout is not retry
A timeout ends one slow attempt. Retry policy is separate and must define attempts, backoff, jitter, idempotency, and the total budget.
async def attempt(deadline: float):
for index in range(3):
try:
async with asyncio.timeout_at(deadline):
return await call()
except TimeoutError:
if index == 2:
raise
await asyncio.sleep(0.1 * (2 ** index))The absolute deadline prevents retries from exceeding the request budget.
Testing timeout behavior
Avoid tests with long real sleeps. Use short but reasonable limits, asyncio.Event, and fakes that complete only after a signal.
async def never(event: asyncio.Event):
await event.wait()
async def test_timeout():
event = asyncio.Event()
try:
async with asyncio.timeout(0.05):
await never(event)
except TimeoutError:
passDo not assert exact millisecond timing in CI. Test behavior: exception, cleanup, and cancellation of related tasks.
Metrics and observability
Distinguish connection timeout, read timeout, queue timeout, operation timeout, and total deadline. Record elapsed time, remaining budget, dependency name, and request ID.
A high timeout rate may indicate saturation, a slow dependency, an overly aggressive deadline, or a blocked event loop.
Common mistakes
- Catching TimeoutError inside the context: conversion occurs on exit.
- Calling time.sleep in async code: the event loop is blocked.
- Swallowing CancelledError: timeout semantics may break.
- Restarting the budget in each layer: prefer absolute deadlines.
- Expecting timeout to kill a thread: async cancellation cannot stop arbitrary native work.
- Using shield without ownership: work may continue unsupervised.
Complete example: shared service budget
import asyncio
async def query_service(name: str, deadline: float) -> dict:
async with asyncio.timeout_at(deadline):
await asyncio.sleep(0.2)
return {"service": name, "ok": True}
async def aggregate() -> list[dict]:
loop = asyncio.get_running_loop()
deadline = loop.time() + 2.0
tasks: list[asyncio.Task[dict]] = []
async with asyncio.timeout_at(deadline):
async with asyncio.TaskGroup() as group:
for name in ["profile", "orders", "balance"]:
tasks.append(
group.create_task(
query_service(name, deadline),
name=f"query-{name}",
)
)
return [task.result() for task in tasks]Every operation shares one deadline and belongs to one structured group. The flow does not create a fresh budget for each dependency.
Conclusion
asyncio.timeout() turns a deadline into a clear scope. It uses cooperative cancellation, converts expiration into TimeoutError, and can protect multiple related awaits.
The official Python asyncio timeout documentation covers timeout(), timeout_at(), rescheduling, and expiration. Use absolute deadlines for shared budgets, preserve cleanup, and treat timeout as one part of a broader resilience policy.
