Set Timeouts and Share One Deadline

Your request handler calls a payment API. Most calls answer in 50 milliseconds, but now and then the API hangs, and the user watches a spinner until their browser gives up. Put a time limit on the call:

import asyncio


async def charge() -> str:
    await asyncio.sleep(10)  # stands in for an API that hangs
    return "charged"


async def main() -> None:
    try:
        async with asyncio.timeout(0.5):
            print(await charge())
    except TimeoutError:
        print("payment API timed out")


asyncio.run(main())

payment API timed out

Catch TimeoutError outside the block

asyncio.timeout(seconds) starts a timer. When it runs out, it cancels the code inside the block, as task.cancel() would. Inside the block, your code sees a CancelledError. As that error leaves the async with, asyncio.timeout replaces it with TimeoutError.

So put the except TimeoutError around the async with, as above. A handler inside the block sees only CancelledError from this timer.

Because the timer works by cancellation, it has the same limits as cancel(). It takes effect only at an await, so blocking code keeps running after the time is up. If the code catches CancelledError and keeps running, the block ends without raising TimeoutError.

An await inside the block can raise a TimeoutError of its own, for example from a database driver. To check whether your timer ran out, keep the timeout in a variable and call expired():

scope = asyncio.timeout(0.5)
try:
    async with scope:
        await charge()
except TimeoutError:
    print("our timer" if scope.expired() else "the driver's own timeout")

Give the whole request one deadline

Now add retries. Say each attempt gets asyncio.timeout(2) and you allow three attempts. A request that promised an answer in two seconds can now take six. Each attempt keeps to its own limit, but no timer covers the total.

Instead, compute one deadline when the request starts, and make every attempt stop at that deadline. loop.time() reads the event loop's clock, and asyncio.timeout_at(when) sets a timer for a time on that clock:

import asyncio

calls = 0


async def flaky_fetch() -> str:
    global calls
    calls += 1
    await asyncio.sleep(0.4)  # each attempt takes 0.4 seconds
    if calls < 3:
        raise ConnectionError("server busy")
    return "page"


async def fetch_with_budget(budget: float) -> str:
    loop = asyncio.get_running_loop()
    deadline = loop.time() + budget
    while loop.time() < deadline:
        try:
            async with asyncio.timeout_at(deadline):
                return await flaky_fetch()
        except ConnectionError:
            print("attempt failed, retrying")
    raise TimeoutError("no time left for another attempt")


async def main() -> None:
    try:
        print(await fetch_with_budget(1.0))
    except TimeoutError:
        print(f"gave up after {calls} attempts")


asyncio.run(main())

attempt failed, retrying
attempt failed, retrying
gave up after 3 attempts

The third attempt starts at 0.8 seconds, and the deadline cancels it at 1.0. With a budget of 2.0, the same code prints page.

Retry only errors that are safe to try again, such as a busy server. A TimeoutError from the deadline means the budget is spent, so let it propagate. Let CancelledError propagate too, because it means someone asked this task to stop.

loop.time() is a monotonic clock, which means it only moves forward and ignores changes to the computer's date and time. Its values mean nothing outside your program. To pass a deadline to another service, send the seconds left.

Nest a tighter limit inside the deadline

You can put a tighter limit on one step inside a larger one:

async with asyncio.timeout_at(deadline):  # the whole request
    user = await load_user()
    try:
        async with asyncio.timeout(0.2):  # recommendations are optional
            recs = await load_recommendations(user)
    except TimeoutError:
        recs = []

If the inner 0.2 seconds run out first, the inner block raises TimeoutError, and the handler falls back to an empty list. If the outer deadline comes first, the inner block lets the cancellation pass unchanged, and the TimeoutError comes out of the outer block. So the inner handler catches only the inner timer's TimeoutError.

Check your understanding You give each of three attempts its own asyncio.timeout(2). How long can one request take? Show answer Hide answer

About six seconds. Each attempt can use its full two seconds, and no timer limits the total. Compute one deadline with loop.time() + budget when the request starts, and give every attempt asyncio.timeout_at(deadline).

Check your understanding Inside async with asyncio.timeout(1):, you wrap the await in try/except TimeoutError. Does that handler run when the timer runs out? Show answer Hide answer

No. Inside the block, the timer raises CancelledError, which except TimeoutError does not catch. asyncio.timeout turns it into TimeoutError only as the error leaves the async with. Put the handler outside the block.

Going further

• asyncio.wait_for(aw, timeout) is the older way to limit a single awaitable, and it still works. asyncio.timeout limits a whole block, so several awaits can share one limit.
• asyncio.timeout(0), or a deadline already in the past, does not raise straight away. The block runs until its next await, and the cancellation happens there.
• A timeout object also has when() and reschedule(). Moving a deadline later breaks the time budget you promised the caller, so do it only on purpose.

See the asyncio timeouts documentation.