Handle Timeout in aiohttp: Set and Manage Request Timeouts

In this tutorial, you’ll learn how to manage timeouts in aiohttp.

You’ll explore various timeout types, configuration methods, and how to handle timeouts in different cases.

aiohttp provides three main types of timeouts: connection, read, and total timeouts. Let’s explore each of them.

 

 

Connection timeout

Connection timeout limits the time spent establishing a connection to the server.

To show a connection timeout, use the following code:

import asyncio
import aiohttp
async def fetch_with_connection_timeout():
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get('http://example.com', timeout=aiohttp.ClientTimeout(connect=0.001)) as response:
                return await response.text()
        except asyncio.TimeoutError:
            return "Connection timeout occurred"
print(asyncio.run(fetch_with_connection_timeout()))

Output:

Connection timeout occurred

This code sets an extremely short connection timeout of 1 millisecond, which will trigger a timeout error.

It shows how connection timeouts work in practice by limiting the time allowed for establishing a connection.

 

Read timeout

Read timeout limits the time waiting for a response from the server after the connection has been established.

To set a read timeout, you should use the sock_read parameter.

import asyncio
import aiohttp
async def fetch_with_read_timeout():
    async with aiohttp.ClientSession() as session:
        try:
            timeout = aiohttp.ClientTimeout(sock_read=2)
            async with session.get('http://httpbin.org/delay/5', timeout=timeout) as response:
                return await response.text()
        except asyncio.TimeoutError:
            return "Read timeout occurred"
print(asyncio.run(fetch_with_read_timeout()))

Output:

Read timeout occurred

The sock_read is used to specify the timeout for reading from the socket.

 

Total timeout

Total timeout limits the overall time for the entire request-response cycle including connection establishment, request sending, and response reading.

Here’s an example of a total timeout:

import asyncio
import aiohttp
async def fetch_with_total_timeout():
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get('http://httpbin.org/delay/3', timeout=aiohttp.ClientTimeout(total=2)) as response:
                return await response.text()
        except asyncio.TimeoutError:
            return "Total timeout occurred"
print(asyncio.run(fetch_with_total_timeout()))

Output:

Total timeout occurred

This code sets a total timeout of 2 seconds for a request to a server that delays its response by 3 seconds.

The total timeout is triggered before the server completes its response.

You can configure specific timeout types individually:

import asyncio
import aiohttp
async def fetch_with_specific_timeouts():
    timeout = aiohttp.ClientTimeout(connect=0.5, sock_read=2, total=None)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.get('http://httpbin.org/delay/1') as response:
                return await response.text()
        except asyncio.TimeoutError:
            return "Timeout occurred"
print(asyncio.run(fetch_with_specific_timeouts()))

Output:

{
  "args": {}, 
  "data": "", 
  "files": {}, 
  "form": {}, 
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Host": "httpbin.org", 
    "User-Agent": "Python/3.11 aiohttp/3.10.5", 
    "X-Amzn-Trace-Id": "Root=1-66dd9af3-2d2b4e19749f0bd033d4a710"
  }, 
  "origin": "197.192.233.95", 
  "url": "http://httpbin.org/delay/1"
}

This code sets specific timeouts for connection and socket read operations, while disabling the total timeout.

The request succeeds because the server’s response time is within the specified limits.

 

Per-session Timeout Settings

You can configure different timeout settings for specific sessions:

import asyncio
import aiohttp
async def fetch_with_different_timeouts():
    short_timeout = aiohttp.ClientTimeout(total=1)
    long_timeout = aiohttp.ClientTimeout(total=10)
    async with aiohttp.ClientSession(timeout=short_timeout) as short_session:
        try:
            async with short_session.get('http://httpbin.org/delay/2') as response:
                short_result = await response.text()
        except asyncio.TimeoutError:
            short_result = "Short timeout occurred"
    print(short_result)
    async with aiohttp.ClientSession(timeout=long_timeout) as long_session:
        async with long_session.get('http://httpbin.org/delay/2') as response:
            long_result = await response.text()
    print(long_result)
asyncio.run(fetch_with_different_timeouts())

Output:

Short timeout occurred
{
  "args": {}, 
  "data": "", 
  "files": {}, 
  "form": {}, 
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Host": "httpbin.org", 
    "User-Agent": "Python/3.11 aiohttp/3.10.5", 
    "X-Amzn-Trace-Id": "Root=1-66dd962a-051a5f002a0454dd692326f3"
  }, 
  "origin": "197.192.233.95", 
  "url": "http://httpbin.org/delay/2"
}

This code shows how to create sessions with different timeout settings.

The first request with a 1-second timeout fails with a timeout message, while the second request with a 10-second timeout succeeds and returns the response.