Timeouts и retries: как пережить сетевые сбои без потери данных

Сеть — это место, где всё ломается. TCP-соединения обрываются, серверы падают, прокси зависают, DNS перестаёт резолвиться, lambda холодно стартует. ETL pipeline без правильных timeout-ов и retry-ев работает на одной и той же машине, но в production падает каждый второй запуск. В этом уроке — фундамент production-ready HTTP-клиента: правильные timeouts, retry-стратегии (fixed/exponential/jitter), и главное — что можно ретраить, а что нельзя.

Timeouts: всегда явный, всегда раздельный

Главное правило: никогда не делай HTTP-запрос без timeout. Дефолт в requests — None (бесконечный). Сервер может зависнуть, и твой ETL будет ждать пока его не убьёт OS killer через час.

requests: timeout — int, float или tuple

import requests

# Один timeout на всё (connect + read)
requests.get(url, timeout=30)            # 30 секунд

# Tuple: (connect_timeout, read_timeout)
requests.get(url, timeout=(5, 30))       # 5с на connect, 30с на read

# Дефолтный для Session -- нет, надо ставить на каждый запрос
# Workaround: подменять request метод

httpx: structured Timeout

import httpx

# Простой
httpx.get(url, timeout=30.0)             # 30с на всё

# Структурированный
timeout = httpx.Timeout(
    connect=5.0,    # TCP+TLS handshake
    read=30.0,      # время на чтение body
    write=10.0,     # время на отправку body
    pool=5.0,       # время на получение connection из pool
)
httpx.get(url, timeout=timeout)

# На клиенте -- дефолт для всех запросов
with httpx.Client(timeout=timeout) as c:
    r = c.get(url)  # использует client-default
    r = c.get(url, timeout=60.0)  # override на конкретный запрос

# Отключить timeout совсем
httpx.get(url, timeout=None)

Best practice timeouts для DE pipeline:

• connect=5s — сервер либо ответит быстро, либо не работает.
• read=30-300s — большой JSON может качаться долго, особенно через slow API типа Salesforce.
• pool=5s — если все pool-соединения заняты дольше 5с, что-то не так.
• write=10s — для upload файлов увеличить до 60-300с.

# Production timeouts для типичного DE pipeline
timeout = httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)

Retry через urllib3.Retry (для requests)

requests использует urllib3 под капотом, и вся retry-логика — там.

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

retry_strategy = Retry(
    total=3,                     # макс retry (всего)
    backoff_factor=0.5,          # delay = backoff_factor * (2 ** attempt) -- exponential
    status_forcelist=[429, 500, 502, 503, 504],  # на каких статусах ретраить
    allowed_methods=["GET", "PUT", "DELETE", "HEAD", "OPTIONS"],  # ТОЛЬКО idempotent
    respect_retry_after_header=True,  # если сервер прислал Retry-After -- слушаемся
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

# Теперь все запросы через session автоматически ретраятся
r = session.get("https://flaky-api.example.com/users/1", timeout=30)

Поведение backoff_factor=0.5:

• attempt 0 -> delay 0s (первый retry без задержки)
• attempt 1 -> delay 0.5 * 2 = 1.0s
• attempt 2 -> delay 0.5 * 4 = 2.0s
• attempt 3 -> delay 0.5 * 8 = 4.0s

В сумме до 4-х retry: 0 + 1 + 2 + 4 = 7 секунд retry-окно. Total=3 значит 3 retry после первой попытки = 4 запроса максимум.

status_forcelist=[429, 500, 502, 503, 504] — стандартный набор. Не ретрай 4xx (кроме 429), они “не пройдут” повторами:

• 429 Too Many Requests — rate limit, retry с backoff правильно.
• 500 Internal Server Error — generic, может быть transient.
• 502 Bad Gateway — upstream проблема, transient.
• 503 Service Unavailable — временно недоступен, transient.
• 504 Gateway Timeout — upstream не ответил, transient.
• 400-403 — клиентская ошибка, retry не поможет.
• 404 — нет такого ресурса, retry не поможет.

Tenacity: универсальный retry для любой логики

urllib3.Retry — только для HTTP requests. Если нужен retry на произвольную функцию (чтение файла, вызов БД, custom RPC) — используй tenacity. Это библиотека для retry с декоратором.

$ pip install tenacity==9.0.0

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)),
)
def fetch_user(user_id):
    r = httpx.get(f"https://api.example.com/users/{user_id}", timeout=30)
    r.raise_for_status()
    return r.json()

Что значат параметры:

Stop conditions

from tenacity import stop_after_attempt, stop_after_delay, stop_any

stop_after_attempt(5)           # макс 5 попыток
stop_after_delay(60)            # макс 60 секунд retry-цикла (включая waits)
stop_any(stop_after_attempt(5), stop_after_delay(60))  # либо то либо другое

Wait strategies

from tenacity import wait_fixed, wait_exponential, wait_random, wait_random_exponential

wait_fixed(2)                    # ровно 2 секунды между попытками
wait_exponential(multiplier=1, min=1, max=60)
# wait = min(max, multiplier * (2 ** attempt))
# attempt 0: 1s, 1: 2s, 2: 4s, 3: 8s, ..., capped at 60s

wait_random(min=1, max=5)        # случайно от 1 до 5 секунд
wait_random_exponential(multiplier=1, max=60)
# exponential + jitter -- лучшая стратегия в production

Retry conditions

from tenacity import retry_if_exception_type, retry_if_result

retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError))
# ретраим только на эти exceptions, остальные пропускаем

retry_if_result(lambda r: r.status_code in (502, 503, 504))
# ретраим если функция вернула что-то с статусом из списка

Полный production пример

from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception_type,
    before_sleep_log,
)
import logging
import httpx

log = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(multiplier=1, max=30),  # exponential + jitter
    retry=retry_if_exception_type((
        httpx.TimeoutException,
        httpx.NetworkError,
        httpx.HTTPStatusError,  # включает 4xx -- фильтруем дальше
    )),
    before_sleep=before_sleep_log(log, logging.WARNING),  # логируем каждый retry
    reraise=True,  # после исчерпания попыток -- кидаем оригинальный exception
)
def fetch_with_retry(url):
    r = httpx.get(url, timeout=30)
    if r.status_code in (429, 500, 502, 503, 504):
        # Эти статусы -- ретраим
        r.raise_for_status()
    elif r.status_code >= 400:
        # 4xx (кроме 429) -- не ретраим, сразу пробрасываем
        raise httpx.HTTPStatusError(
            f"Non-retryable: {r.status_code}", request=r.request, response=r
        )
    return r.json()

Детальнее:

• wait_random_exponential — exponential backoff с jitter (рандомизация чтобы клиенты не синхронизировались — thundering herd).
• before_sleep_log — каждый retry логируется.
• reraise=True — после 5 попыток кидает последний exception (без обёртки в RetryError).

Jitter — почему это критично

Без jitter: 1000 клиентов одновременно получают 503 от сервера, все ретраят через 1s, через 2s, через 4s — синхронно. Сервер получает 1000 запросов одновременно каждые 2-4-8 секунд — никогда не восстановится. Это thundering herd.

С jitter: те же 1000 клиентов ретраят через random(0, 1), random(0, 2), random(0, 4) — равномерно размазано во времени. Сервер получает по 100-200 запросов в секунду — справляется, восстанавливается.

# Без jitter (плохо при больших флотах клиентов)
wait=wait_exponential(multiplier=1, max=60)
# attempt 0: 1s, 1: 2s, 2: 4s -- все клиенты в одну точку

# С jitter (production-grade)
wait=wait_random_exponential(multiplier=1, max=60)
# attempt 0: random(0, 1), 1: random(0, 2), 2: random(0, 4)

AWS, Google, Azure SDK — все используют jittered exponential backoff. Это must для облачных API.

Idempotency и retry: что можно ретраить

Сценарий риска: твой код шлёт POST /orders, сервер успешно создал заказ, но ответ потерян в сети (TCP-reset, прокси-таймаут). Клиент видит ConnectionError, retry — создаётся ВТОРОЙ заказ. Двойное списание денег.

Защита: генерируй Idempotency-Key UUID на одну логическую операцию, шли в header. Сервер дедуплицирует. См. урок 04-rest-design/02-idempotency.

import uuid
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
def safe_post(client, url, data):
    # Ключ генерируем ВНЕ retry -- переиспользуем на всех попытках
    pass

def create_order(client, order_data):
    idempotency_key = str(uuid.uuid4())  # один раз

    @retry(stop=stop_after_attempt(3), wait=wait_exponential())
    def _attempt():
        return client.post(
            "/orders",
            json=order_data,
            headers={"Idempotency-Key": idempotency_key},  # переиспользуем
        )

    return _attempt()

Retry-After header

Некоторые API (особенно с rate limit 429) шлют header Retry-After — сколько секунд подождать перед повтором.

HTTP/1.1 429 Too Many Requests
Retry-After: 30

urllib3.Retry(respect_retry_after_header=True) — слушается. tenacity — нужно реализовать вручную:

import httpx
import time

def fetch_respecting_retry_after(url, max_attempts=5):
    for attempt in range(max_attempts):
        r = httpx.get(url)
        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", 30))
            print(f"Rate limited, sleeping {wait}s")
            time.sleep(wait)
            continue
        r.raise_for_status()
        return r.json()
    raise RuntimeError("Max retries exceeded")

Полный production HTTP-клиент

import httpx
import uuid
from tenacity import (
    retry, stop_after_attempt, wait_random_exponential,
    retry_if_exception_type, before_sleep_log,
)
import logging

log = logging.getLogger(__name__)


class APIClient:
    def __init__(self, base_url, token):
        self.client = httpx.Client(
            base_url=base_url,
            headers={"Authorization": f"Bearer {token}"},
            timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0),
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
            http2=True,
        )

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.client.close()

    @retry(
        stop=stop_after_attempt(5),
        wait=wait_random_exponential(multiplier=1, max=30),
        retry=retry_if_exception_type((
            httpx.TimeoutException,
            httpx.NetworkError,
        )),
        before_sleep=before_sleep_log(log, logging.WARNING),
        reraise=True,
    )
    def get(self, path, **kwargs):
        r = self.client.get(path, **kwargs)
        if r.status_code in (429, 500, 502, 503, 504):
            r.raise_for_status()  # вызовет HTTPStatusError, не сматчится в retry
        elif r.status_code >= 400:
            r.raise_for_status()  # non-retryable, пробрасываем
        return r

    def post(self, path, json=None, **kwargs):
        # POST с Idempotency-Key для безопасного retry
        key = str(uuid.uuid4())
        kwargs.setdefault("headers", {})["Idempotency-Key"] = key

        @retry(
            stop=stop_after_attempt(3),
            wait=wait_random_exponential(multiplier=1, max=10),
            retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)),
            reraise=True,
        )
        def _attempt():
            r = self.client.post(path, json=json, **kwargs)
            r.raise_for_status()
            return r
        return _attempt()


# Использование
with APIClient("https://api.example.com", "my_token") as api:
    user = api.get("/users/1").json()
    new_order = api.post("/orders", json={"user_id": 1, "items": [...]}).json()

Попробуй сам

import httpx
from tenacity import retry, stop_after_attempt, wait_random_exponential

# httpbin.org/status/<code> вернёт статус, который мы попросим
# Проверим retry на 503

attempts = []

@retry(
    stop=stop_after_attempt(4),
    wait=wait_random_exponential(multiplier=1, max=4),
    reraise=True,
)
def fetch():
    attempts.append(1)
    print(f"Attempt {len(attempts)}")
    r = httpx.get("https://httpbin.org/status/503", timeout=10)
    r.raise_for_status()
    return r.json()

try:
    fetch()
except httpx.HTTPStatusError as e:
    print(f"Failed after {len(attempts)} attempts: {e}")
# Attempt 1
# (wait random)
# Attempt 2
# ...
# Failed after 4 attempts: ...

Поэкспериментируй: меняй stop_after_attempt, wait_random_exponential(max=...). Замеряй сколько времени занимает retry-цикл с разными параметрами.

Killer takeaway

Timeouts: ВСЕГДА явные. requests: timeout=30 или tuple (connect, read). httpx: httpx.Timeout(connect, read, write, pool) — четыре фазы отдельно. Best practice: connect=5s, read=30-60s для DE pipeline. Retry для requests: urllib3.Retry через HTTPAdapter — total, backoff_factor, status_forcelist=[429, 500, 502, 503, 504], allowed_methods= только idempotent. Retry универсальный: tenacity декоратор с stop_after_attempt, wait_random_exponential (jitter обязателен в production — иначе thundering herd), retry_if_exception_type. Что можно ретраить: GET/HEAD/OPTIONS/PUT/DELETE — да. POST/PATCH — только с Idempotency-Key (UUID, генерируется ОДИН раз вне retry, переиспользуется на всех попытках). Иначе — дубли.