Асинхронная обработка списка заказов и связанных данных о пользователях

Условие

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

1. По order_id отправляет запрос к orders_service_url и получает идентификатор user_id.

2. С полученным user_id одновременно делает два запроса:

   • к user_service_url — сервис возвращает JSON‑объект с информацией о пользователе;
   • к af_service_url — сервис возвращает JSON‑объект с данными о доставке.

3. После получения ответов от обоих сервисов передаёт их в функцию process_users_and_shipments(user: dict, af_info: dict) -> Any для дальнейшей бизнес‑логики.

from typing import Any

# Сначала в order_service смотрим, какой заказ делал какой пользователь
orders_service_url = "http://some-domain.ru/api/v1/user-orders"

# С этими пользователями идем в:
user_service_url = "http://some-domain.ru/api/v1/user"
af_service_url = "http://some-domain.ru/api/v1/af"

# 10 миллионов запросов
# В качестве результата на order service будет возвращен id пользователя user_id
order_requests = [
    {"order_id": 25},
    {"order_id": 26},
    {"order_id": 34},
    # ...
]

def process_users_and_shipments(user: dict, af_info: dict) -> Any:
    # some business logic
    ...

Функция должна быть реализована с использованием асинхронных средств (например, asyncio и aiohttp) для того, чтобы эффективно обрабатывать большое количество запросов без блокировки.

Ответ

После получения user_id запустите два независимых запроса конкурентно. Для миллионов заказов используйте ограниченный пул работников и потоковый вход, а не миллионы созданных Task.

Ключевая операция для одного заказа в Python 3.11+:

import asyncio

async def process_order(order, get_json):
    info = await get_json(orders_service_url, {'order_id': order['order_id']})
    user_id = info['user_id']
    async with asyncio.TaskGroup() as group:
        user_task = group.create_task(get_json(user_service_url, {'user_id': user_id}))
        af_task = group.create_task(get_json(af_service_url, {'user_id': user_id}))
    return process_users_and_shipments(user_task.result(), af_task.result())

get_json должен использовать общий async HTTP-клиент, timeout, проверку статуса и формата. Запускайте фиксированное число workers, получающих заказы из ограниченной asyncio.Queue; producer читает источник порциями. Результаты сохраняйте постепенно. Тяжёлую синхронную обработку вынесите из event loop. Нужны checkpoint, ограниченные повторы и идемпотентность, иначе перезапуск продублирует бизнес-действия.