Learn Timeouts with asyncio.wait_for()

Swipe to show menu

Network requests can hang indefinitely. A server may be slow, unresponsive, or down. Without a timeout, your program waits forever. asyncio.wait_for() lets you set a maximum time for any awaitable to complete.

Basic Usage

asyncio.wait_for(coroutine, timeout) runs the coroutine and raises asyncio.TimeoutError if it doesn't complete within timeout seconds.


              123456789101112131415161718192021222324
            
import asyncio
import httpx
import nest_asyncio

nest_asyncio.apply()

# Fetching a post with a strict timeout
async def fetch_post(client, post_id):
    url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
    response = await client.get(url)
    return response.json()["title"]

async def main():
    async with httpx.AsyncClient() as client:
        try:
            title = await asyncio.wait_for(
                fetch_post(client, 1),
                timeout=5.0,
            )
            print(f"Title: {title}")
        except asyncio.TimeoutError:
            print("Request timed out")

asyncio.run(main())

Simulating a Timeout

httpbin.org/delay/{n} delays the response by n seconds – useful for testing timeout behavior:


              1234567891011121314151617181920212223
            
import asyncio
import httpx
import nest_asyncio

nest_asyncio.apply()

# Triggering a timeout with a slow endpoint
async def fetch_slow(client):
    response = await client.get("https://httpbin.org/delay/5")
    return response.json()

async def main():
    async with httpx.AsyncClient() as client:
        try:
            result = await asyncio.wait_for(
                fetch_slow(client),
                timeout=2.0,  # Times out before the 5s delay completes
            )
            print(result)
        except asyncio.TimeoutError:
            print("Timed out after 2 seconds")

asyncio.run(main())

Applying Timeouts to Multiple Tasks

Wrap each task individually to control timeouts per request:


              123456789101112131415161718192021222324252627282930313233
            
import asyncio
import httpx
import nest_asyncio

nest_asyncio.apply()

# Fetching a post with a specific timeout
async def fetch_with_timeout(client, post_id, timeout):
    url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
    try:
        # Waiting for the response within the given timeout
        response = await asyncio.wait_for(
            client.get(url),
            timeout=timeout,
        )
        return response.json()["title"]
    except asyncio.TimeoutError:
        return f"Post {post_id} timed out"

async def main():
    # Creating an asynchronous client session
    async with httpx.AsyncClient() as client:
        # Gathering results using different timeout values
        results = await asyncio.gather(
            fetch_with_timeout(client, 1, timeout=1.0),
            fetch_with_timeout(client, 2, timeout=2.0),
            fetch_with_timeout(client, 3, timeout=5.0),
        )

    for result in results:
        print(result)

asyncio.run(main())

What Happens on Timeout

When wait_for() times out:

It cancels the wrapped coroutine immediately;
It raises asyncio.TimeoutError to the caller;
The cancelled coroutine receives CancelledError at its next await point.

This means resources held by the coroutine (open connections, file handles) are properly cleaned up if the coroutine handles CancelledError correctly.

Everything was clear?

Thanks for your feedback!

Section 2. Chapter 4

Ask AI

Ask anything or try one of the suggested questions to begin our chat

Section 2. Chapter 4