Self-Task: APIs, JSON, and Asynchronous Python

Create a new public repository or a separate project folder. The assignment does not require branches or code from previous homework.

Imagine you have a list of users in a JSON file and you need to get more information about each user through an API (web service). This is a very common task in real work.

Basic Concepts

1. JSON file

JSON is a text data exchange format. Example:

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

• JSON is widely used to exchange structured data via HTTP API.
• uuid — a unique user identifier within the task.

2. API (web service)

• The API describes the contract for interaction between programs. In this task, the client receives data from an HTTP service.
• The client must be able to make GET requests to an address like: {base_url}/users/{uuid}. base_url passed through an argument or configuration, rather than being hard-wired into business logic.
• The API returns different responses:
• 200: everything is fine, data received
• 400: request error
• 404: user not found

api.example.com The following is used as an example contract only. A real external service is not needed to complete the task: HTTP responses are reproduced in tests via aioresponses.

3. Asynchronous programming

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()

• Asynchronous code allows you to concurrently wait for multiple network requests without blocking the thread on each wait.
• async declares a coroutine function, and await suspends the current coroutine until the awaitable operation completes.
• One ClientSession should be reused for a series of requests: it controls the connection pool.

Decorators

Decorator is a function that changes the behavior of another function. In our case, it will add the current time to the response from the service and do a retry only for selected temporary errors: network failures, 408, 429 taking into account Retry-After and suitable 5xx. Majority 4xx there is no point in repeating.

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

What needs to be implemented?

1. Main function:

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

2. Decorator with parameters:

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

Testing

Why are tests needed?

1. Check that the code works correctly
2. Helps find errors
3. Document how the code should work

What will we test?

1. Read/Write JSON:

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

2. Working with API (using aioresponses):

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

3. Decorator (using freezegun): freezegun can be used to capture a calendar date and time as a result. Measure the duration of execution using monotonous clocks (time.monotonic()), and in the test replace this particular time source.

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

Libraries that we will use

1. aiohttp - for HTTP requests
2. pytest - to run tests
3. aioresponses - for mocking HTTP requests in tests
4. freezegun - for working with time in tests

What to pay attention to

1. Error handling - what to do if the file does not open?
2. Logging - record important events
3. Retries - what to do if the request fails?
4. Tests should be clear and test important cases

Additional task

1. Use the library click to run your script with configuration from the command line, like this:

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

Readiness criteria

• the project is launched according to the instructions from the README;
• input JSON is validated, and read errors and data structures are handled;
• one ClientSession reused for a series of queries;
• requests are executed concurrently with timeout and concurrency limits;
• retry applies only to selected temporary errors and has a clear number of retries;
• duration is measured by monotonous hours;
• tests are independent of the availability of an external API and check 200, 404, temporary error, retry exhaustion and invalid JSON;
• the result is saved in the output JSON, and the running and testing commands are described in the README.

Official sources

• asyncio: tasks and coroutines
• json in the standard library
• aiohttp client quickstart
• pytest documentation