Проект: API-клиент на requests с таймаутами, повторами и кэшем

Девять уроков мы набирали детали: первый GET, параметры, статусы и raise_for_status, три лица ответа, POST-формы, заголовки, сессии с куками, авторизацию. По отдельности это детали. Сегодня собираем их в механизм, который переживёт контакт с настоящим миром: класс WeatherClient — обёртка над погодным API с таймаутом, умными повторами и кэшем. Именно такие клиенты живут в продакшене: бот дёргает их из aiogram, дашборд — из Django, скрипт — из cron.

Сеть в браузерной песочнице нет — и для проекта это даже удобнее: мы построим клиента поверх консервированных ответов, записанных JSON-литералов, которые ведут себя как настоящие. Вся логика — таймауты, счётчики повторов, кэш — настоящая и исполняется прямо на странице. Поменяешь транспорт на requests — получишь боевого клиента, он идёт в конце урока целиком.

Что строим: клиент погоды без сети

Тестировщики называют консервированные ответы фикстурами: это снимки реальных ответов API, сохранённые в коде. Транспорт клиента просит данные — фикстуры их отдают, притворяясь сетью. Логика клиента при этом не знает подмены: она получает те же словари, что получила бы от живого сервера.

import json

# Консервированные ответы: то, что вернул бы реальный API
CANNED = {
    "Москва": {"temp": 18, "wind": 3, "desc": "облачно"},
    "Сочи": {"temp": 26, "wind": 1, "desc": "ясно"},
    "Мурманск": {"temp": 7, "wind": 9, "desc": "ветрено"},
}


class APIError(Exception):
    """Своя ошибка API: отличаем её от случайных сбоев Python."""


print(sorted(CANNED))

try:
    raise APIError("нет данных для города: Атлантида")
except APIError as e:
    print("поймали APIError:", e)

['Москва', 'Мурманск', 'Сочи']
поймали APIError: нет данных для города: Атлантида

Первый кирпич — APIError, собственный класс ошибки. Зачем он, если Python и так умеет сотню исключений? Затем, что вызывающему коду важно ловить одним `except APIError` все проблемы сервиса — а сбои самого Python (TypeError, KeyError в твоём коде) не маскировать под них. CANNED — обычный словарь, но заметь: его значения выглядят ровно как ответ API, поля temp, wind, desc. Извлечём из фикстуры человеческий отчёт — это и есть работа клиентского метода:

import json

# Ответ, который "прислал" сервер: строка JSON
raw = '{"temp": 18, "wind": 3, "desc": "облачно"}'
data = json.loads(raw)

report = (
    f"Погода: {data['desc']}, {data['temp']} градусов,"
    f" ветер {data['wind']} м/с"
)
print(report)
print("полей в ответе:", len(data))

Погода: облачно, 18 градусов, ветер 3 м/с
полей в ответе: 3

Таймаут: почему без него нельзя

Первый механизм клиента — timeout, предел ожидания одного запроса в секундах. Без него сценарий фатален: сервер, который молчит, подвесит твой скрипт навсегда — requests по умолчанию готов ждать бесконечно. Ночной скрипт с бесконечным ожиданием в три часа ночи останавливает всю выгрузку, а с таймаутом — честно получает исключение через пять секунд и переходит к повтору.

import requests

url = "https://api.example.com/v1/data"

# Без таймаута зависший сервер подвесит скрипт навсегда
resp = requests.get(url, timeout=5)          # 5 секунд на всё

# Точнее: пара (на соединение, на чтение ответа)
resp = requests.get(url, timeout=(3, 10))

Таймаут ничего не лечит — он лишь гарантирует, что зависание закончится исключением через N секунд. А вот что делать с этим исключением — следующий механизм.

Повторы: экспоненциальная задержка

Сети дёшево ломаются на секунду: просел Wi-Fi, сервер перезапустился, балансировщик переключил ноду. Поэтому клиент повторяет неудачные запросы. Но повторять нужно с паузами и растущими паузами — экспоненциальной задержкой: полсекунды, секунда, две. Если сервис прилёг на минуту, три попытки с растущими паузами дают ему время подняться — и не превращаются в пинг-понг из десятков запросов. Посчитаем расписание повторов без всякой сети:

# Экспоненциальная задержка: считаем, а не спим
MAX_RETRIES = 3
BASE_DELAY = 0.5

delays = [BASE_DELAY * 2**n for n in range(MAX_RETRIES)]
print("задержки, сек:", delays)
print("худший случай, сек:", sum(delays))

for attempt in range(1, MAX_RETRIES + 1):
    print("попытка", attempt, "из", MAX_RETRIES)

задержки, сек: [0.5, 1.0, 2.0]
худший случай, сек: 3.5
попытка 1 из 3
попытка 2 из 3
попытка 3 из 3

Формула одна: BASE_DELAY * 2**n. Три попытки — три задержки — 3.5 секунды худшего случая между первым запросом и капитуляцией. В настоящем коде между попытками стоит time.sleep(delay); в наших фикстурах спать незачем — счётчик попыток и так честный, а страница не заставляет тебя ждать. Капитулировать клиент тоже должен красиво: выбросить свою ошибку с понятным текстом, а не traceback десятой строки чужой библиотеки.

import time

CACHE_TTL = 600                          # ответ живёт 10 минут

cache = {}                               # город -> (момент, ответ)


def from_cache(city):
    if city not in cache:
        return None
    born, data = cache[city]
    if time.monotonic() - born > CACHE_TTL:
        del cache[city]                  # просрочен: идём в сеть
        return None
    return data

Кэш: словарь, который экономит запросы

Третий механизм — самый простой и самый прибыльный. Дашборд спрашивает погоду в Москве каждые пять минут, а меняется она раз в час — девять из десяти запросов можно вообще не отправлять, отдавая прежний ответ из словаря. Кэш в нашем клиенте — словарь: ключ — город, значение — последний ответ. Второй и следующие запросы того же города обслуживаются мгновенно и без сети.

import json


class APIError(Exception):
    """Базовая ошибка нашего API-клиента."""


class ServerError(APIError):
    """5xx или таймаут: сервер не справился, повтор имеет смысл."""


class NotFound(APIError):
    """404: города нет, повторять бессмысленно."""


CANNED = {
    "Москва": {"temp": 18, "wind": 3, "desc": "облачно"},
    "Сочи": {"temp": 26, "wind": 1, "desc": "ясно"},
}


class FakeTransport:
    """Заглушка сети: дважды ломается, потом отвечает как настоящий API."""

    def __init__(self):
        self.failures_left = 2
        self.calls = 0

    def get(self, city):
        self.calls += 1
        if city not in CANNED:
            raise NotFound(f"в базе нет города {city}")
        if self.failures_left > 0:
            self.failures_left -= 1
            raise ServerError("таймаут 5 секунд")
        return CANNED[city]


class WeatherClient:
    def __init__(self, transport, max_retries=3):
        self.transport = transport
        self.max_retries = max_retries
        self.cache = {}
        self.cache_hits = 0

    def get_current(self, city):
        # Шаг 1: кэш - на тот же вопрос отвечаем без запроса
        if city in self.cache:
            self.cache_hits += 1
            return self.cache[city]
        # Шаг 2: повторы - только для ServerError (5xx и таймауты)
        last_error = None
        for attempt in range(1, self.max_retries + 1):
            try:
                data = self.transport.get(city)
            except NotFound:
                raise                       # 404 повторами не лечится
            except ServerError as e:
                last_error = e
                print("  попытка", attempt, "не удалась:", e)
            else:
                self.cache[city] = data
                return data
        raise ServerError(f"повторы исчерпаны: {last_error}")


client = WeatherClient(FakeTransport())

w = client.get_current("Москва")
print("Москва:", w["desc"] + ",", w["temp"], "градусов")

w = client.get_current("Москва")
print("Москва снова:", w["desc"], "- обращений к сети не было")

try:
    client.get_current("Тверь")
except NotFound as e:
    print("Тверь:", e)

print("обращений к серверу:", client.transport.calls)
print("попаданий в кэш:", client.cache_hits)

  попытка 1 не удалась: таймаут 5 секунд
  попытка 2 не удалась: таймаут 5 секунд
Москва: облачно, 18 градусов
Москва снова: облачно - обращений к сети не было
Тверь: в базе нет города Тверь
обращений к серверу: 4
попаданий в кэш: 1

Разбери вывод построчно — это вся логика клиента в шести строках. Первый запрос к Москве: транспорт дважды симулирует таймаут (смотри строки про попытки 1 и 2), на третьей попытке отвечает — данные уходят в кэш. Повторный вопрос о Москве: кэш отработал мгновенно, счётчик попаданий вырос, транспорт даже не потревожили. Запрос о Твери: NotFound пролетел насквозь без единого повтора — правило «4xx не лечатся» в действии. Итог: четыре обращения к серверу на три вызова клиента — без кэша было бы пять, а на большом дашборде разница превращается в тысячи сэкономленных запросов.

Полный клиент на requests: файл целиком

Механика проверена — меняем транспорт с фикстур на настоящие requests. Обрати внимание на структуру файла: она такая же строгая, как в нашем мини-клиенте, и это неслучайно — хорошая структура клиента читается сверху вниз как оглавление.

# weather_client.py - карта файла настоящего клиента
# 1. Импорты: requests и time - больше ничего не нужно
# 2. Константы: BASE_URL, TIMEOUT, MAX_RETRIES, BASE_DELAY
# 3. Исключения: APIError и NotFound - свой язык ошибок
# 4. Класс WeatherClient:
#    __init__     - Session, User-Agent, пустой кэш
#    get_current  - кэш, повторы, разбор статусов
# 5. Внизу файла - использование: client = WeatherClient() ...

import time
import requests

# ---------- 1. Константы: вся настройка в одном месте ----------
BASE_URL = "https://api.open-meteo.com/v1/forecast"
TIMEOUT = 5            # секунд на один запрос
MAX_RETRIES = 3        # всего попыток
BASE_DELAY = 0.5       # первая задержка перед повтором, сек


# ---------- 2. Свои исключения ----------
class APIError(Exception):
    """Наша ошибка API: ловится одним except."""


class NotFound(APIError):
    """404: данных нет, повторять бессмысленно."""


# ---------- 3. Класс клиента ----------
class WeatherClient:
    def __init__(self, session=None):
        # Session из урока 8: общие заголовки и keep-alive
        self.session = session or requests.Session()
        self.session.headers["User-Agent"] = "WeatherClient/1.0"
        self._cache = {}               # ключ -> ответ

    def get_current(self, city, lat, lon):
        key = (city, lat, lon)
        if key in self._cache:         # кэш: тот же вопрос без сети
            return self._cache[key]

        params = {
            "latitude": lat,
            "longitude": lon,
            "current_weather": "true",
        }
        last_error = None
        for attempt in range(1, MAX_RETRIES + 1):
            try:
                resp = self.session.get(
                    BASE_URL, params=params, timeout=TIMEOUT
                )
            except (requests.exceptions.Timeout,
                    requests.exceptions.ConnectionError) as e:
                last_error = e         # сеть или таймаут: повторяем
            else:
                if resp.status_code == 404:
                    raise NotFound(f"нет данных для {city}")
                if resp.status_code >= 500:
                    last_error = APIError(f"сервер: {resp.status_code}")
                elif resp.ok:
                    self._cache[key] = resp.json()
                    return self._cache[key]
                else:
                    raise APIError(f"наша ошибка: {resp.status_code}")
            time.sleep(BASE_DELAY * 2 ** (attempt - 1))  # 0.5, 1, 2 сек

        raise APIError(f"повторы исчерпаны: {last_error}")


# ---------- 4. Использование ----------
client = WeatherClient()
weather = client.get_current("Москва", 55.75, 37.62)
print(weather["current_weather"]["temperature"])

Сравни с фикстурной версией — совпадает всё, кроме транспорта: вместо transport.get() — session.get(..., timeout=TIMEOUT), вместо ServerError от заглушки — честные Timeout, ConnectionError и статусы 5xx. Именно ради такого совпадения мы строили мини-клиента: поменял одну деталь — и проект из учебного стал рабочим.

Чек-лист: клиент готов к продакшену?

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

1. timeout стоит на каждом запросе — скрипт не может зависнуть навсегда;
2. повторы включены только для таймаутов, ошибок сети и 5xx — 4xx поднимаются наверх сразу;
3. задержка между повторами экспоненциальная, попыток не больше трёх-пяти;
4. свой класс ошибок: вызывающий код ловит один except APIError, а не десяток чужих;
5. кэш или хотя бы уважение к лимитам API — ключ не выжигается дашбордом за утро;
6. секреты (ключи, токены) читаются из переменных окружения, в коде и git их нет;
7. User-Agent назван честно — серверы видят, кто к ним приходит.

Наш WeatherClient проходит все семь пунктов. Мелким шрифтом: у кэша здесь нет срока жизни — для дашборда с погодой добавь в ключ время или чисть словарь по таймеру, но это уже домашняя работа, каркас у тебя есть.

Итоги курса: что ты теперь умеешь

Десять уроков назад GET-запрос выглядел как строка из чужой документации. Теперь ты прочитываешь HTTP-обмен целиком: строишь запрос с параметрами и честным User-Agent, различаешь коды ответа, выбираешь между text, content и json, скачиваешь файлы, отправляешь формы и JSON, держишь куки в Session, представляешься Basic и Bearer — и упаковываешь всё это в клиент, который не рассыпается от плохой сети. requests остаётся в любом проекте — от парсера на BeautifulSoup до бота на aiogram.

Куда свернуть дальше? Если понравилось разобрать HTTP по байтам — в финальном проекте курса про регулярные выражения так же по косточкам разбирают логи. Если хочется держать состояние и общаться с людьми, а не с серверами — пройди проект телеграм-бота на aiogram: там твои новые знания про HTTP, сессии и авторизацию зарабатывают сообщения в реальном чате. А пока — сохрани weather_client.py: это каркас, который ты будешь копировать в свои проекты ещё годы.

Хороший клиент — как хороший официант: принял заказ с таймаутом на ожидание, повторил его вежливо и принёс то, что просили, или честно сказал, что кухни нет.