Управление таймаутами HTTP запросов средствами Python

Расширенные примеры настройки таймаута

Пример 1. Таймаут с повторными попытками и обработкой ошибок

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time

session = requests.Session()
retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retries)
session.mount('http://', adapter)
session.mount('https://', adapter)

urls = ['https://httpbin.org/delay/1', 'https://httpbin.org/delay/6']
for url in urls:
    try:
        start = time.time()
        # Таймаут 5 секунд для всего запроса
        response = session.get(url, timeout=5)
        elapsed = time.time() - start
        print(f'{url} - статус {response.status_code}, время {elapsed:.2f}s')
    except requests.exceptions.Timeout:
        print(f'{url} - превышен таймаут')
    except requests.exceptions.ConnectionError:
        print(f'{url} - ошибка соединения')

https://httpbin.org/delay/1 - статус 200, время 1.23s
https://httpbin.org/delay/6 - превышен таймаут

Пример 2. Асинхронный таймаут с aiohttp и разными лимитами

import asyncio
import aiohttp

async def fetch_with_timeout(url, timeout_total=10, timeout_connect=5):
    timeout = aiohttp.ClientTimeout(total=timeout_total, connect=timeout_connect)
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(url, timeout=timeout) as resp:
                return await resp.json()
        except asyncio.TimeoutError as e:
            return {'error': str(e)}

async def main():
    result = await fetch_with_timeout('https://httpbin.org/delay/8', timeout_total=8)
    print(result)

asyncio.run(main())

{'error': ''}

В данном примере общий таймаут 8 секунд, и запрос к delay/8 успеет вернуть ответ до истечения. Если увеличить задержку, будет выброшено исключение.

Пример 3. Использование Eventlet для таймаута без изменения кода

import eventlet
eventlet.monkey_patch()
import requests

# Устанавливаем таймаут с помощью примитивов Eventlet
with eventlet.Timeout(3, False):  # False значит не выбрасывать исключение, а просто вернуть None
    try:
        response = requests.get('https://httpbin.org/delay/4')
        print(response.status_code)
    except eventlet.Timeout:
        print('Таймаут по Eventlet')

Таймаут по Eventlet

Обратите внимание: eventlet.Timeout с аргументом False не генерирует исключение, а просто позволяет выполнению продолжиться. Для обработки таймаута нужно явно проверять возвращаемое значение.

Пример 4. Таймаут на уровне Transport Adapter в urllib3

import requests
from requests.packages.urllib3 import PoolManager, Timeout

class CustomAdapter(requests.adapters.HTTPAdapter):
    def send(self, request, **kwargs):
        kwargs['timeout'] = Timeout(connect=2, read=10)
        return super().send(request, **kwargs)

session = requests.Session()
session.mount('https://', CustomAdapter())

response = session.get('https://httpbin.org/delay/3')
print(response.status_code)

Этот метод позволяет тонко настроить таймауты, переопределив метод send.

200

Пример 5. Комбинирование таймаута с кэшированием и повторными попытками в одном адаптере

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class TimeoutRetryAdapter(HTTPAdapter):
    def __init__(self, timeout=(5, 10), retries=3):
        self.timeout = timeout
        self.max_retries = Retry(total=retries, backoff_factor=1)
        super().__init__(max_retries=self.max_retries)

    def send(self, request, **kwargs):
        kwargs.setdefault('timeout', self.timeout)
        return super().send(request, **kwargs)

session = requests.Session()
adapter = TimeoutRetryAdapter(timeout=(3, 8), retries=2)
session.mount('https://', adapter)

response = session.get('https://httpbin.org/delay/2')
print(response.status_code)

200