Самостоятельное задание: API, JSON и асинхронный Python

Создайте новый публичный репозиторий или отдельную папку проекта. Задание не требует веток и кода из предыдущих домашних работ.

Представьте, что у вас есть список пользователей в JSON файле, и вам нужно получить дополнительную информацию о каждом пользователе через API (веб-сервис). Это очень частая задача в реальной работе.

Основные понятия

1. JSON файл

JSON — текстовый формат обмена данными. Пример:

{
    "users": [
        {"name": "Иван", "uuid": "123-456"},
        {"name": "Мария", "uuid": "789-012"}
    ]
}

JSON широко используется для обмена структурированными данными через HTTP API.
uuid — уникальный идентификатор пользователя в рамках задачи.

2. API (веб-сервис)

API описывает контракт взаимодействия программ. В этой задаче клиент получает данные от HTTP-сервиса.
Клиент должен уметь делать GET-запросы по адресу вида: {base_url}/users/{uuid}. base_url передаётся через аргумент или конфигурацию, а не зашивается в бизнес-логику.
API возвращает разные ответы:
200: всё хорошо, данные получены
400: ошибка в запросе
404: пользователь не найден

api.example.com ниже используется только как пример контракта. Реальный внешний сервис для выполнения задания не нужен: HTTP-ответы воспроизводятся в тестах через aioresponses.

3. Асинхронное программирование

async def get_user_info(session: aiohttp.ClientSession, base_url: str, uuid: str):
    async with session.get(f"{base_url}/users/{uuid}") as response:
        response.raise_for_status()
        return await response.json()

Асинхронный код позволяет конкурентно ожидать несколько сетевых запросов, не блокируя поток на каждом ожидании.
async объявляет coroutine function, а await приостанавливает текущую корутину до завершения awaitable-операции.
Одну ClientSession следует переиспользовать для серии запросов: она управляет connection pool.

Декораторы

Декоратор - это функция, которая изменяет поведение другой функции. В нашем случае он будет добавлять текущее время в ответ от сервиса и делать retry только для выбранных временных ошибок: сетевых сбоев, 408, 429 с учётом Retry-After и подходящих 5xx. Большинство 4xx повторять бессмысленно.

@async_retry(attempts=3)  # Это декоратор
async def my_function():
    # Если функция упадёт с ошибкой, декоратор попробует ещё 2 раза
    pass

Что нужно реализовать?

Основную функцию:

async def process_users(input_file: str, output_file: str):
    # 1. Читаем JSON
    # 2. Создаём одну ClientSession
    # 3. Запускаем запросы конкурентно через TaskGroup/gather
    #    и ограничиваем concurrency и timeout
    # 4. Сохраняем результаты
    pass

Декоратор с параметрами:

def async_retry(attempts: int = 3):
    # Измеряет время работы
    # Пытается повторить при ошибке
    # Логирует результаты
    pass

Тестирование

Почему нужны тесты?

Проверяют, что код работает правильно
Помогают найти ошибки
Документируют, как должен работать код

Что будем тестировать?

Чтение/запись JSON:

def test_read_json():
    data = read_json("test.json")
    assert "users" in data

Работу с API (используя aioresponses):

async def test_api_call():
    with aioresponses() as mock:
        mock.get('http://api.example.com/users/123',
                status=200,
                payload={"status": "active"})

Декоратор (используя freezegun): freezegun можно использовать, чтобы зафиксировать календарные дату и время в результате. Длительность выполнения измеряйте монотонными часами (time.monotonic()), а в тесте подменяйте именно этот источник времени.

@freeze_time("2024-01-01")
async def test_retry():
    # Проверяем, что retry работает
    # Проверяем измерение времени
    pass

Библиотеки, которые будем использовать

aiohttp - для HTTP запросов
pytest - для запуска тестов
aioresponses - для мока HTTP запросов в тестах
freezegun - для работы со временем в тестах

На что обратить внимание

Обработка ошибок - что делать, если файл не открылся?
Логирование - записывать важные события
Повторные попытки - что делать, если запрос не удался?
Тесты должны быть понятными и проверять важные случаи

Дополнительное задание

Используйте библиотеку click для запуска вашего скрипта с конфигурацией из командной строки, например:

python -m script --input file.json --output result.json

Критерии готовности

проект запускается по инструкции из README;
входной JSON валидируется, а ошибки чтения и структуры данных обрабатываются;
одна ClientSession переиспользуется для серии запросов;
запросы выполняются конкурентно с timeout и ограничением concurrency;
retry применяется только к выбранным временным ошибкам и имеет понятное число попыток;
длительность измеряется монотонными часами;
тесты не зависят от доступности внешнего API и проверяют 200, 404, временную ошибку, исчерпание retry и некорректный JSON;
результат сохраняется в выходной JSON, а команды запуска и тестирования описаны в README.

Unit tests