Handle Timeouts and Retries in aiohttp

Network requests fail. Servers go down. Connections drop. If you build async apps with aiohttp, you must plan for these problems.

Timeouts and retries are your safety net. They keep your app stable when the network is not.

This guide shows you how to handle timeouts and retries in aiohttp. It is written for beginners. Every concept comes with clear code and output.

Why Timeouts Matter

A request without a timeout can hang forever. Your app waits. Your users wait. Nothing happens.

Timeouts fix this. They set a limit on how long a request can take. When the limit is reached, aiohttp raises an error. You can then handle it.

Always set a timeout. Never rely on the default behavior for production code.

Setting a Timeout in aiohttp

The simplest way is to pass a timeout value to your request. The value is in seconds.

import aiohttp
import asyncio

async def fetch():
    async with aiohttp.ClientSession() as session:
        # Timeout after 5 seconds
        async with session.get("https://httpbin.org/delay/10", timeout=5) as resp:
            return await resp.text()

asyncio.run(fetch())

This request asks the server to wait 10 seconds. But our timeout is 5 seconds. So aiohttp stops the request early.

asyncio.exceptions.TimeoutError

The request is cancelled. Your app does not hang. This is the core idea behind timeouts.

Using ClientTimeout for Fine Control

For more control, use the ClientTimeout class. It lets you set different limits for different phases of a request.

You can set a total timeout. You can also set a connect timeout. And a sock read timeout.

import aiohttp
import asyncio

async def fetch():
    timeout = aiohttp.ClientTimeout(total=10, connect=3, sock_read=5)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get("https://httpbin.org/get") as resp:
            return await resp.json()

asyncio.run(fetch())

Here is what each value means:

total is the whole request time. connect is the time to open a connection. sock_read is the time to read data from the socket.

This setup is great for APIs. You can fail fast on slow connections but allow more time for big responses.

Catching Timeout Errors

When a timeout happens, aiohttp raises asyncio.TimeoutError. You should catch it and decide what to do.

import aiohttp
import asyncio

async def fetch():
    timeout = aiohttp.ClientTimeout(total=3)
    try:
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.get("https://httpbin.org/delay/5") as resp:
                return await resp.text()
    except asyncio.TimeoutError:
        print("Request timed out. Try again later.")
        return None

asyncio.run(fetch())

Request timed out. Try again later.

This pattern is simple and safe. Your app keeps running even when the network fails.

Why You Need Retries

Some failures are temporary. A server may be busy. A connection may drop for a second. A retry can fix these problems.

Retries are not about ignoring errors. They are about giving a request a second chance.

But be careful. Not every error should be retried. A 404 error will not fix itself. A timeout might.

Adding Retries with a Simple Loop

You can build a retry loop yourself. It is easy to understand and control.

import aiohttp
import asyncio

async def fetch_with_retry(url, retries=3):
    timeout = aiohttp.ClientTimeout(total=5)
    for attempt in range(1, retries + 1):
        try:
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.get(url) as resp:
                    return await resp.text()
        except (asyncio.TimeoutError, aiohttp.ClientError) as e:
            print(f"Attempt {attempt} failed: {e}")
            if attempt == retries:
                print("All retries failed.")
                return None
            await asyncio.sleep(2 ** attempt)  # exponential backoff

asyncio.run(fetch_with_retry("https://httpbin.org/delay/10"))

Attempt 1 failed: 
Attempt 2 failed: 
Attempt 3 failed: 
All retries failed.

This code tries three times. It waits longer after each failure. This is called exponential backoff. It gives the server time to recover.

Using aiohttp-retry for Cleaner Code

Writing retry logic by hand works. But a library can make it cleaner. The aiohttp-retry package adds retries to your session with little code.

import aiohttp
import asyncio
from aiohttp_retry import RetryClient, ExponentialRetry

async def fetch():
    retry_options = ExponentialRetry(attempts=3)
    async with RetryClient(retry_options=retry_options) as client:
        async with client.get("https://httpbin.org/delay/10") as resp:
            return await resp.text()

asyncio.run(fetch())

This does the same job as the manual loop. But it is shorter and easier to read.

The library also supports status-based retries. You can retry only on 500 errors or 429 rate limits.

Best Practices for Timeouts and Retries

Follow these rules to keep your app healthy.

Always set a timeout. A missing timeout is a hidden bug.

Use exponential backoff. Do not hammer a failing server. Wait longer each time.

Limit retry attempts. Three to five tries is usually enough. More can slow your app.

Retry only safe requests. GET requests are safe to retry. POST requests may create duplicate data.

Log your failures. Logs help you spot patterns and fix real problems.

Combining Timeouts and Retries

Timeouts and retries work best together. A timeout stops a slow request. A retry gives it another chance.

import aiohttp
import asyncio

async def fetch_with_retry(url, retries=3):
    timeout = aiohttp.ClientTimeout(total=5)
    for attempt in range(1, retries + 1):
        try:
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.get(url) as resp:
                    resp.raise_for_status()
                    return await resp.json()
        except (asyncio.TimeoutError, aiohttp.ClientError) as e:
            print(f"Attempt {attempt} failed: {type(e).__name__}")
            if attempt == retries:
                return None
            await asyncio.sleep(2 ** attempt)

async def main():
    data = await fetch_with_retry("https://httpbin.org/get")
    print("Success:", data is not None)

asyncio.run(main())

Success: True

This pattern is a solid base for most async apps. It handles slow servers and short outages.

Common Mistakes to Avoid

Many developers make the same errors. Avoid these traps.

Do not retry on 400 errors. These are client errors. Retrying will not help.

Do not skip the timeout. A retry loop without timeouts can hang forever.

Do not retry too fast. Add a delay between attempts. Use backoff.

Do not create a new session for every request. Reuse one session for better performance.

Conclusion

Timeouts and retries are essential for reliable aiohttp apps. Timeouts stop slow requests. Retries recover from short failures.

Start with a simple timeout. Add a retry loop with backoff. Then consider a library like aiohttp-retry for cleaner code.

Keep your limits small. Log your failures. Test your error paths. With these habits, your async apps will stay strong even when the network is not.