Внешний мир ненадёжен

В уроке 03 мы научились вытаскивать из API всё, страница за страницей. Но на 537-й странице из 600 у вас сеть моргнула и httpx.ReadTimeout обрушил весь job. Перезапустили — теперь GitHub вернул 502 Bad Gateway. Ещё перезапустили — 429 Too Many Requests, потому что предыдущие фейлы успели сжечь квоту.

Это не редкие катастрофы — это штатная работа с внешними API. Junior, который надеется на «обычно работает», в production пишет очень хрупкие пайплайны. Junior, который сразу обкладывает каждый вызов API правильной retry-логикой, пишет boring и надёжно.

В этом уроке — какие ошибки повторять, как именно повторять, и как делегировать всю эту работу библиотеке tenacity, чтобы не писать свой @retry (как мы делали в уроке про декораторы — для понимания, а не для прода).

Какие ошибки повторять, какие — нет

Сначала классификация — не каждая ошибка стоит повтора.

Можно повторять — transient errors (временные):

• 5xx от сервера (500, 502, 503, 504) — сервер сломался, повторение может сработать.
• 429 Too Many Requests — превысили лимит, надо подождать.
• Network errors — httpx.ConnectError, httpx.ReadTimeout, httpx.RemoteProtocolError. Сеть моргнула.
• DNS-фейлы — тоже бывают временными.

НЕ повторять — permanent errors (постоянные):

• 4xx кроме 429 — это наша ошибка. 400 — мусор послали, 401 — токена нет, 403 — прав нет, 404 — ресурса нет. Повторять бессмысленно: ситуация не изменится через секунду.
• JSONDecodeError, ValidationError — данные не такие, как мы ждали. Повтор даст те же данные.
• KeyError, AttributeError — баг у нас.

Условно — зависит от метода:

• POST без Idempotency-Key — НЕ повторять автоматически (см. урок 01). Может задвоить данные.
• POST с Idempotency-Key — можно.
• GET, PUT, DELETE — идемпотентные по протоколу, можно.

Это правило — фильтр входа в retry-логику. Любая прод-ready обвязка первым делом проверяет тип ошибки и решает: повторять или сразу пробрасывать наверх.

Exponential backoff и jitter

Если запрос упал — глупо сразу повторять, скорее всего получите тот же ответ. Пауза перед повтором даёт серверу шанс прийти в себя.

attempt 1 → fail → wait 1s
attempt 2 → fail → wait 2s
attempt 3 → fail → wait 4s
attempt 4 → fail → wait 8s
attempt 5 → fail → wait 16s

Это даёт серверу всё больше времени между нашими попытками — и одновременно ограничивает суммарное время (за 6 попыток — 31 секунда, не часы).

Но есть второй эффект, важный, когда вы — не единственный клиент. Представьте, что упал общий downstream и 1000 клиентов одновременно вошли в backoff. Они все ждут ровно 1, 2, 4, 8 секунд — и все стучатся в сервер в одни и те же моменты. Сервер только-только встал — на него падает thundering herd, он снова умирает.

Решение —

В прод-коде никогда не используйте чистый exponential backoff без jitter. Это известный footgun.

tenacity: библиотека для retry

Своими декораторами @retry можно писать. Промышленный инструмент —

uv add tenacity

Простейший пример:

from tenacity import retry, stop_after_attempt, wait_exponential
import httpx


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=16),
)
def fetch(url: str) -> dict:
    response = httpx.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

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

• stop=stop_after_attempt(5) — максимум 5 попыток.
• wait=wait_exponential(multiplier=1, min=1, max=16) — задержка 1, 2, 4, 8, 16 (потолок 16).

По умолчанию tenacity повторяет на любом исключении. Это слишком грубо: ловить 404 и пробовать ещё 4 раза — пустая трата времени и квоты.

Условный retry — retry_if_exception_type

Фильтр на тип ошибки:

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type,
)
import httpx


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=16, jitter=2),
    retry=retry_if_exception_type((
        httpx.ConnectError,
        httpx.ReadTimeout,
        httpx.RemoteProtocolError,
    )),
    reraise=True,
)
def fetch(url: str) -> dict:
    response = httpx.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

Что добавилось:

• wait_exponential_jitter — backoff + jitter. initial=1, max=16, jitter=2 означает базовая задержка 1, 2, 4, 8, 16 плюс random добавка от 0 до 2 секунд.
• retry=retry_if_exception_type(...) — повторяем только на сетевых ошибках. На HTTPStatusError (которое кидает raise_for_status на 4xx/5xx) этот retry не сработает — это мы добавим следующим шагом.
• reraise=True — если retry’и исчерпались, кинуть исходную ошибку, не RetryError-обёртку. Удобнее ловить.

Retry на основе результата — retry_if_result

HTTPStatusError объединяет все 4xx/5xx в один тип. Но повторять надо только 5xx и 429, а не 404. Один из способов — retry_if_exception с предикатом:

from tenacity import retry_if_exception


def is_retriable_http_error(exc: BaseException) -> bool:
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in {429, 500, 502, 503, 504}
    return isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout, httpx.RemoteProtocolError))


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=16, jitter=2),
    retry=retry_if_exception(is_retriable_http_error),
    reraise=True,
)
def fetch(url: str) -> dict:
    response = httpx.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

Теперь:

• 200 OK → возврат без retry.
• 404 → HTTPStatusError, не retriable, пробрасывается сразу.
• 503 → retriable, попробуем ещё.
• ReadTimeout → retriable.
• JSONDecodeError → не retriable, пробросится наверх.

Это уже близко к production-grade retry.

429 и Retry-After

Особый случай — 429 Too Many Requests. Сервер часто говорит в заголовке Retry-After, сколько именно ждать:

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

или с указанием HTTP-даты:

Retry-After: Wed, 21 Oct 2026 07:28:00 GMT

Уважать Retry-After — это хороший тон и часто требование API (Stripe откровенно блокирует на день, если игнорировать).

В tenacity это можно сделать через свой wait callback:

from tenacity import wait_base
from email.utils import parsedate_to_datetime
from datetime import datetime, UTC


class wait_retry_after(wait_base):
    """Уважает Retry-After заголовок, если он есть; иначе — экспонента с jitter."""

    def __init__(self, fallback: wait_base) -> None:
        self._fallback = fallback

    def __call__(self, retry_state) -> float:
        exc = retry_state.outcome.exception() if retry_state.outcome else None
        if isinstance(exc, httpx.HTTPStatusError):
            ra = exc.response.headers.get("retry-after")
            if ra:
                try:
                    return float(ra)
                except ValueError:
                    target = parsedate_to_datetime(ra)
                    return max(0.0, (target - datetime.now(UTC)).total_seconds())
        return self._fallback(retry_state)


@retry(
    stop=stop_after_attempt(5),
    wait=wait_retry_after(wait_exponential_jitter(initial=1, max=16, jitter=2)),
    retry=retry_if_exception(is_retriable_http_error),
    reraise=True,
)
def fetch(url: str) -> dict:
    response = httpx.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

Теперь:

• Если сервер сказал Retry-After: 30 — ждём ровно 30 секунд, не экспоненту.
• Если сетевая ошибка / нет Retry-After — обычный exponential backoff с jitter.

Hooks: логирование retry

Перед каждым retry полезно лог в structlog (вспоминаем урок 03-06) — иначе ваши retry будут немой статистикой:

import structlog

log = structlog.get_logger()


def log_before_retry(retry_state) -> None:
    log.warning(
        "retrying",
        attempt=retry_state.attempt_number,
        fn=retry_state.fn.__name__,
        exc=str(retry_state.outcome.exception()),
        wait=retry_state.next_action.sleep if retry_state.next_action else None,
    )


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=16),
    retry=retry_if_exception(is_retriable_http_error),
    before_sleep=log_before_retry,
    reraise=True,
)
def fetch(url: str) -> dict:
    ...

before_sleep — callback, который зовётся перед каждой паузой между попытками. В production такие логи дают понимать: «GitHub API за ночь дал 1200 ретраев на 502» — повод писать в чат с поддержкой.

Rate limiting на стороне клиента

tenacity решает проблему реактивно: ждём ошибку, потом подстраиваемся. Но если у API лимит 5000 в час, а у вас 10 параллельных воркеров, можно сжечь квоту за пять минут — будет каскад 429.

Лучше проактивно ограничивать скорость со стороны клиента. Простейший инструмент —

from pyrate_limiter import Duration, Rate, Limiter

# Не больше 80 запросов в минуту (запас от лимита 100)
limiter = Limiter(Rate(80, Duration.MINUTE))


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=16),
    retry=retry_if_exception(is_retriable_http_error),
)
def fetch(url: str) -> dict:
    limiter.try_acquire("github-api")    # блокируется, если квота кончилась
    response = httpx.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

Это страховка от того, что ваш код сжигает квоту. Внешние факторы (другие клиенты с тем же токеном) она не закроет — это уже история про координацию.

Для DE-проектов с одним токеном на один Airflow-job обычно достаточно лимита немного ниже официального и retry на 429 как страховка от случайных всплесков.

DE-кейс: production-ready клиент

Соберём всё вместе. Это шаблон, который ляжет в основу любого DE-клиента к REST API:

import httpx
import structlog
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception,
)
from collections.abc import Iterator
from typing import Any
import re

log = structlog.get_logger()


def is_retriable(exc: BaseException) -> bool:
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in {429, 500, 502, 503, 504}
    return isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout, httpx.RemoteProtocolError))


class GitHubClient:
    def __init__(self, token: str) -> None:
        self._client = httpx.Client(
            base_url="https://api.github.com",
            headers={
                "Authorization": f"Bearer {token}",
                "Accept": "application/vnd.github+json",
                "User-Agent": "my-etl/1.0",
            },
            timeout=httpx.Timeout(30.0, connect=5.0),
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "GitHubClient":
        return self

    def __exit__(self, *args: object) -> None:
        self.close()

    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential_jitter(initial=1, max=16, jitter=2),
        retry=retry_if_exception(is_retriable),
        before_sleep=lambda s: log.warning(
            "retrying",
            attempt=s.attempt_number,
            exc=str(s.outcome.exception()),
        ),
        reraise=True,
    )
    def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
        response = self._client.request(method, path, **kwargs)
        response.raise_for_status()
        return response

    @staticmethod
    def _parse_link(link: str) -> dict[str, str]:
        result: dict[str, str] = {}
        for part in link.split(","):
            match = re.match(r'\s*<([^>]+)>\s*;\s*rel="([^"]+)"', part)
            if match:
                result[match.group(2)] = match.group(1)
        return result

    def paginate(self, path: str, *, params: dict[str, Any] | None = None) -> Iterator[dict[str, Any]]:
        response = self._request("GET", path, params=params)
        yield from response.json()
        while link := response.headers.get("link"):
            links = self._parse_link(link)
            if "next" not in links:
                return
            response = self._request("GET", links["next"])
            yield from response.json()

В paginate retry применяется к каждой странице отдельно. Если 537-я страница упала с 503 — повторим её, не весь job. Это даёт пайплайнам ту самую устойчивость, ради которой мы всё это пишем.

Что мы получили

• Retriable: 5xx, 429, network errors. Permanent: 4xx (кроме 429), JSONDecodeError, баги. POST — только с Idempotency-Key.
• Exponential backoff + jitter — стандарт. Без jitter — thundering herd на восстанавливающемся сервере.
• tenacity — production-grade библиотека для retry. stop, wait, retry, before_sleep, reraise=True.
• Retry-After — уважать обязательно. Custom wait_base подкласс.
• Rate limiting на клиенте через pyrate-limiter — страховка от сжигания квоты.
• Hooks (before_sleep) — логируйте retry, иначе они потеряются в проде.

Это конец «внешний мир по HTTP». Со следующего урока — другой внешний мир: PostgreSQL.