Работа с API в Python: авторизация, лимиты и повторы

Ваш скрипт тянет данные из чужого API. Первые сто запросов проходят, на сто первом прилетает 429 Too Many Requests, и дальше сыплются ошибки, потому что вы бьёте в закрытую дверь. Или другой сценарий: токен живёт час, а выгрузка идёт полтора, и посреди пагинации сервер отвечает 401. Оба случая — не про плохой сервер, а про то, что клиент не умеет отступать и обновляться. Разберём, как это чинится в Python: где хранить и обновлять токен, как читать заголовки лимитов и как настроить повторы, чтобы одна сетевая икота не убивала всю задачу.

Статья про клиентскую сторону на requests 2.31.x и httpx 0.27.x. Дизайн самого API, вебхуки и кеширование ответов — отдельные темы, здесь их не трогаем.

Авторизация: три схемы и где хранить секрет

Большинство HTTP API используют одну из трёх схем. Отличаются они тем, что клиент кладёт в запрос и как часто это протухает.

Ключ и пароль не хардкодят в исходнике. Держите их в переменных окружения или в секрет-менеджере, а в коде читайте через os.environ. Так секрет не утечёт в git и не всплывёт в логах CI.

Сессия с постоянным заголовком

Если у вас статический ключ, не подставляйте заголовок в каждый вызов руками. Заведите одну сессию — она переиспользует TCP-соединения и хранит общие заголовки.

import os
import requests

# requests 2.31.x
API_KEY = os.environ["SERVICE_API_KEY"]  # заглушка: значение приходит из окружения
BASE_URL = "https://api.example.com/v1"

session = requests.Session()
session.headers.update({"Authorization": f"Bearer {API_KEY}"})

resp = session.get(f"{BASE_URL}/orders", timeout=10)
resp.raise_for_status()  # бросит исключение на 4xx/5xx
data = resp.json()

Обратите внимание на timeout=10. Без него запрос по умолчанию ждёт бесконечно: если сервер завис, ваш скрипт зависнет вместе с ним. Таймаут ставьте всегда.

OAuth2: обновление токена по 401

С OAuth2 access-токен протухает. Наивный подход — обновлять его по расписанию — ломается, если часы клиента и сервера разошлись или сервер отозвал токен раньше. Надёжнее реагировать на 401: получили — обновили токен один раз и повторили запрос.

import os
import time
import requests

TOKEN_URL = "https://api.example.com/oauth/token"
BASE_URL = "https://api.example.com/v1"


class ApiClient:
    def __init__(self):
        self.session = requests.Session()
        self._token = None
        self._expires_at = 0.0

    def _fetch_token(self):
        resp = self.session.post(
            TOKEN_URL,
            data={
                "grant_type": "client_credentials",
                "client_id": os.environ["CLIENT_ID"],
                "client_secret": os.environ["CLIENT_SECRET"],
            },
            timeout=10,
        )
        resp.raise_for_status()
        payload = resp.json()
        self._token = payload["access_token"]
        # обновим за 30 секунд до реального истечения, чтобы не ловить гонку
        self._expires_at = time.monotonic() + payload["expires_in"] - 30

    def _auth_header(self):
        if self._token is None or time.monotonic() >= self._expires_at:
            self._fetch_token()
        return {"Authorization": f"Bearer {self._token}"}

    def get(self, path):
        resp = self.session.get(
            f"{BASE_URL}{path}", headers=self._auth_header(), timeout=10
        )
        if resp.status_code == 401:
            # токен мог протухнуть раньше срока: обновляем и пробуем один раз
            self._fetch_token()
            resp = self.session.get(
                f"{BASE_URL}{path}", headers=self._auth_header(), timeout=10
            )
        resp.raise_for_status()
        return resp.json()

Ключевая деталь — time.monotonic() вместо time.time(). Монотонные часы не прыгают при синхронизации NTP, поэтому расчёт «истёк или нет» не сломается, если системное время подвели.

Лимиты: читайте заголовки, а не гадайте

Сервер почти всегда сообщает, сколько запросов вам ещё осталось. Стандарта на имена нет, но чаще всего это X-RateLimit-Limit, X-RateLimit-Remaining и X-RateLimit-Reset. Точные имена смотрите в документации вашего API — они различаются.

Главный заголовок при 429 — Retry-After. В нём либо число секунд, либо HTTP-дата. Сервер прямо говорит, когда можно вернуться. Игнорировать это и долбить в цикле — быстрый способ получить бан по IP.

Клиент, который не читает Retry-After и повторяет запрос сразу, не ускоряет работу. Он превращает временный лимит в устойчивую очередь ошибок и заставляет сервер тратить ресурсы на отказы вам же.

Проактивно можно притормаживать, пока Remaining не упал до нуля:

def respect_rate_limit(resp):
    """Пауза, если сервер сообщил, что лимит почти исчерпан."""
    remaining = resp.headers.get("X-RateLimit-Remaining")
    reset = resp.headers.get("X-RateLimit-Reset")
    if remaining is not None and reset is not None:
        if int(remaining) == 0:
            import time
            # Reset обычно в секундах до сброса окна; сверьтесь с докой API
            time.sleep(int(reset))

Повторы: что повторять, а что нет

Повторять можно не всё. Разберём по кодам ответа.

• 429 и 503 — повторяем, уважая Retry-After. Это временная перегрузка.
• 500, 502, 504 — повторяем с задержкой. Это сбой на стороне сервера или прокси, часто разовый.
• Сетевые ошибки (ConnectionError, таймаут чтения) — повторяем.
• 400, 401, 403, 404, 422 — не повторяем. Тело запроса или права не изменятся от повтора, вы просто зря нагрузите сервер.

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

Задержка: почему нельзя фиксированную

Если все клиенты после 503 ждут ровно 1 секунду и бьют одновременно, они снова обрушат сервер синхронной волной. Отсюда два правила: экспоненциальный рост задержки и случайный разброс (jitter). Формула паузы перед попыткой номер n: base * 2 ** n плюс случайная добавка.

Не пишите ретраи руками — используйте urllib3

В requests повторы делает встроенный в urllib3 класс Retry через транспортный адаптер. Он сам считает бэкофф и читает Retry-After.

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

# requests 2.31.x поверх urllib3 2.x
retry = Retry(
    total=5,                       # всего до 5 повторов
    backoff_factor=0.5,            # паузы 0.5, 1, 2, 4... секунды
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "PUT", "DELETE"],  # POST намеренно исключён
    respect_retry_after_header=True,
)

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

resp = session.get("https://api.example.com/v1/orders", timeout=10)
resp.raise_for_status()

Параметр называется allowed_methods в urllib3 1.26 и новее; в старых версиях он назывался method_whitelist и сейчас удалён. Если ловите TypeError на этом аргументе — у вас древняя urllib3, обновите.

В httpx встроенного ретрая по статус-кодам нет: транспорт умеет повторять только на уровне соединения (httpx.HTTPTransport(retries=...)), но не по 429. Для полноценной логики берут внешнюю библиотеку, например tenacity, и оборачивают вызов декоратором. Точный список параметров tenacity сверяйте с её документацией — API там менялся между мажорными версиями.

Если не сработало

• Повторы «не видны» — вы смонтировали адаптер на https://, а бьёте в http:// (или наоборот). Смонтируйте оба, как в примере выше.
• Скрипт висит и не падает — забыли timeout. Без него зависший сервер держит поток вечно, и никакой Retry не поможет, потому что попытка просто не завершается.
• Повторяется даже 404 — вы вручную ловите исключение и повторяете в своём цикле, игнорируя код ответа. Проверяйте status_code перед повтором.
• После обновления токена всё равно 401 — токен обновился, но старый заголовок остался на session.headers. Передавайте заголовок в сам вызов, а не только в сессию, либо перезаписывайте его в сессии после каждого обновления.
• Дубли заказов после сбоя — повторяется POST без ключа идемпотентности. Уберите POST из allowed_methods или передавайте Idempotency-Key.

Что настроить дальше

Первое — включите структурное логирование: пишите код ответа, значение Retry-After и номер попытки. Без этого вы не отличите «сервер медленный» от «мы забанены». Второе — вынесите базовый URL, таймаут и параметры Retry в конфиг, чтобы не менять их в пяти местах. Третье — если вызовов много и они независимы, посмотрите в сторону асинхронного httpx с ограничителем параллельности (httpx.Limits): это отдельная тема, но именно она снимает потолок в производительности клиента.