HTTP и API: requests, таймауты, ретраи, пагинация

Половина источников у инженера данных это не база, а чей-то API: CRM, рекламный кабинет, платёжный шлюз, внутренний сервис. Код requests.get(url).json() работает на демо и ломается в проде первой же ночью: сервис ответил 503, подвис на минуту, вернул не все страницы. Урок про то, чем запрос в пайплайне отличается от запроса в консоли.

Запрос, который не повиснет

import requests
 
response = requests.get(
    "https://api.example.com/orders",
    params={"since": "2026-06-01", "limit": 100},
    headers={"Authorization": f"Bearer {token}"},
    timeout=(5, 30),
)
response.raise_for_status()
data = response.json()

Четыре вещи, без которых запрос нельзя пускать в расписание.

timeout обязателен. Без него requests ждёт ответа вечно, и задача в Airflow висит, пока её не убьют руками. Кортеж это таймаут на соединение и на чтение отдельно: соединение должно происходить быстро, чтение может занять дольше.

raise_for_status() превращает статусы 4xx и 5xx в исключение HTTPError. Без него response.json() на странице с текстом «Service Unavailable» упадёт с невнятным JSONDecodeError, а если сервер вернул JSON с описанием ошибки, вы молча запишете его в витрину как данные.

params вместо ручной сборки адреса: requests сам закодирует пробелы и кириллицу. Токен в заголовке, а не в адресе: адреса попадают в логи прокси и серверов.

session = requests.Session() вместо requests.get при многих запросах к одному хосту: сессия держит соединение открытым и общие заголовки, что экономит на установке TLS сотни миллисекунд на каждом запросе.

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

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

Повтор на 400 это три одинаковых ошибки в логе вместо одной. Повтор на 401 это бан токена за перебор.

Пауза между повторами растёт: 1, 2, 4, 8 секунд (экспоненциальный backoff), плюс случайная добавка, чтобы сто ваших задач после сбоя сервиса не ударили по нему одновременно в одну секунду. Число попыток ограничено: три-пять, потом задача падает и будит дежурного.

import random
import time
 
RETRY_STATUSES = {429, 500, 502, 503, 504}
 
def get_with_retry(session, url, attempts=4, **kwargs):
    for attempt in range(attempts):
        try:
            response = session.get(url, timeout=(5, 30), **kwargs)
        except (requests.ConnectionError, requests.Timeout):
            if attempt == attempts - 1:
                raise
        else:
            if response.status_code not in RETRY_STATUSES:
                response.raise_for_status()
                return response
            if attempt == attempts - 1:
                response.raise_for_status()
        time.sleep(2 ** attempt + random.random())

Это тот же retry из урока про декораторы, только с логикой по статусу. В requests можно вместо этого подключить urllib3.Retry к сессии через HTTPAdapter, он делает то же самое, включая Retry-After.

Пагинация

API не отдаёт миллион записей одним ответом. Три схемы пагинации, от худшей к лучшей.

page=1, 2, 3 (по номеру страницы): просто, но если между запросами добавилась запись, страницы сдвинутся, и одна запись придёт дважды, а другая потеряется.

offset и limit: то же самое, плюс на больших смещениях база на стороне API работает всё медленнее.

Курсор: ответ содержит токен next, который нужно передать в следующий запрос; когда его нет, страницы кончились. Записи не сдвигаются, потому что курсор привязан к позиции, а не к номеру.

def fetch_all(session, url):
    cursor = None
    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor
        page = get_with_retry(session, url, params=params).json()
        yield from page["items"]
        cursor = page.get("next")
        if not cursor:
            return

Генератор отдаёт записи по мере получения страниц: обработка и запись в базу идут параллельно с выгрузкой, в памяти одна страница. Защита от бесконечного цикла, если API вернёт тот же курсор дважды, стоит в коде недорого: проверка, что курсор изменился, или лимит страниц.

Инкремент и идемпотентность

Выгружать всё каждый раз нельзя: долго и дорого для источника. Выгружают изменения с прошлого запуска: since=<время последней успешной выгрузки>. Время берётся из состояния, которое сохраняется после успешной записи в витрину, а не до. Если сохранить его перед записью и упасть на записи, следующий запуск пропустит данные.

Окно берут с запасом назад (например, на час), потому что записи в источнике могут появляться с задержкой и с временем «в прошлом». Запас даёт дубли на стыке окон, и их убирает идемпотентная запись: upsert по ключу записи, как в уроке SQL про повторную загрузку. Инкремент без идемпотентной записи это задвоение.

Для записи в чужой API (отправка событий, создание сущностей) идемпотентность обеспечивает ключ запроса Idempotency-Key: повтор запроса с тем же ключом не создаёт вторую сущность. Если API его поддерживает, ретраи на POST безопасны; если нет, повторять POST нельзя без проверки, прошёл ли первый.

Данные из ответа

Ответ API это чужие данные, и к ним применимо всё из урока про файлы: проверить поля, привести типы, не доверять, что amount число. Схему ответа стоит зафиксировать моделью (dataclass или pydantic) с явным списком полей: когда поставщик переименует поле, вы получите понятную ошибку на входе, а не None в витрине через месяц.

Для API с большим объёмом (рекламные кабинеты, аналитика) есть асинхронные отчёты: вы заказываете отчёт, получаете job_id, опрашиваете статус, скачиваете файл. Опрос это тот же цикл с паузой и лимитом попыток, что и ретраи.

Как отвечать на собеседовании

Как забирать данные из API в пайплайне? Сессия с таймаутом и raise_for_status, повторы с растущей паузой только на сетевые ошибки, 429 и 5xx, пагинация курсором через генератор, инкремент по времени последней успешной выгрузки с запасом и идемпотентной записью.

Какие ошибки повторять, а какие нет? Повторять таймауты, обрывы, 429 и 5xx. Не повторять 4xx кроме 429: проблема в запросе или правах, повтор её не исправит.

Чем курсорная пагинация лучше постраничной? Записи не сдвигаются при изменениях между запросами, и источнику не нужно считать смещение. Признак конца это отсутствие next.

Что такое идемпотентность запроса? Повтор даёт тот же результат, что один вызов. Для чтения это бесплатно, для записи нужен ключ идемпотентности или проверка перед повтором.

Практика

Сети в песочнице нет, поэтому вместо requests в задачах класс FakeSession: он отдаёт заранее заданные ответы со статусами, умеет «падать» первые N раз и отдаёт страницы с курсором. Интерфейс как у настоящей сессии: get(url, params=...) возвращает объект с status_code, json() и raise_for_status().