Name: Sobes.tech
Rating: 4.9 (250 reviews)

Паттерн Waiter для API реализуется путем периодического опроса (polling) ресурса до достижения определенного целевого состояния или до истечения таймаута.

Основные компоненты реализации:

Цель ожидания: Определение условия, при котором ожидание считается завершенным (например, статус ресурса стал "готовым", значение поля достигло определенного значения).
Период опроса (Polling interval): Время между последовательными запросами к API.
Таймаут (Timeout): Максимальное время, в течение которого Waiter будет выполнять опросы.
Логика опроса: Функция или метод, который выполняет GET-запрос к API для получения текущего состояния ресурса.
Проверка состояния: Логика, которая анализирует ответ от API и проверяет, достигнуто ли целевое состояние.
Механизм ожидания: Реализация цикла, который выполняет опросы с заданным интервалом, проверяет состояние и завершается при достижении цели, таймаута или ошибке.

Пример реализации на Python:

import time
import requests
from typing import Dict, Any, Optional, Callable

def wait_until(
    url: str,
    api_key: str,
    condition: Callable[[Dict[str, Any]], bool],
    timeout: int = 60,
    polling_interval: int = 5
) -> Dict[str, Any]:
    """
    Ожидает, пока ресурс по заданному URL не удовлетворит условию.

    Args:
        url: URL ресурса API.
        api_key: Ключ API для аутентификации.
        condition: Функция, принимающая ответ (словарь) и возвращающая True, если условие выполнено.
        timeout: Максимальное время ожидания в секундах.
        polling_interval: Интервал между запросами в секундах.

    Returns:
        Последний полученный ответ от API, когда условие было выполнено.

    Raises:
        TimeoutError: Если условие не было выполнено в течение заданного таймаута.
        requests.exceptions.RequestException: Если произошла ошибка HTTP-запроса.
    """
    start_time = time.time()
    headers = {"X-API-Key": api_key} # Пример заголовка для API ключа

    while time.time() - start_time < timeout:
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status() # Вызывает исключение для плохих статусов (4xx или 5xx)
            data = response.json()

            if condition(data):
                return data # Условие выполнено

        except requests.exceptions.RequestException as e:
            print(f"Ошибка запроса: {e}")
            # Можно добавить логику повторных попыток или выхода

        time.sleep(polling_interval) # Ожидаем перед следующим опросом

    raise TimeoutError(f"Ожидание условия для {url} истекло ({timeout} секунд)")

# Пример использования:
# Предположим, API возвращает статус обработки в поле 'status'
# и мы ждем, пока статус станет 'COMPLETED'
# resource_url = "https://api.example.com/processing_job/123"
# my_api_key = "YOUR_API_KEY"

# def is_completed(data: Dict[str, Any]) -> bool:
#     return data.get("status") == "COMPLETED"

# try:
#     completed_resource_data = wait_until(
#         resource_url,
#         my_api_key,
#         is_completed,
#         timeout=120,
#         polling_interval=10
#     )
#     print("Ресурс готов:", completed_resource_data)
# except (TimeoutError, requests.exceptions.RequestException) as e:
#     print("Не удалось дождаться готовности ресурса:", e)

Особенности реализации:

Обработка ошибок: Необходимо обрабатывать ошибки HTTP (неверные статусы, проблемы с сетью).
Экспоненциальная задержка: Для снижения нагрузки на API можно увеличивать интервал опроса с каждой неудачной попыткой (exponential backoff).
Отмена ожидания: Предусмотреть возможность прерывания ожидания.
Логирование: Добавить логирование для отслеживания прогресса и возникающих проблем.
Параметризация: Сделать Waiter гибким, позволяя настраивать URL, метод запроса (GET, POST), заголовки, параметры, условия ожидания.

Как реализовать паттерн Waiter для API?