Запрашивание миллиона адресов

Условие

1. Дано массив из 1 000 000 URL‑ов, каждый из которых возвращает набор item_id.
2. Для каждого полученного item_id требуется выполнить запросы к трём отдельным сервисам: service1, service2 и service3.
3. Ответы от этих сервисов обрабатываются функцией business_logic, формирующей окончательный результат.
4. Все полученные результаты собираются в финальный список.

from typing import List

# Примеры входных данных:
# формат: url, request_data
requests_samples = [
    ('http://some-service/getItems/', {'user_id': 100}),
    ('http://some-service/getItems/', {'user_id': 101}),
    ...
]

# Сервисы для опроса
service_1_url = 'http://service1/fillItems/'
service_2_url = 'http://service2/scoreItems/'
service_3_url = 'http://service3/logItems/'

# Функция для обработки ответов сервисов
def business_logic(service1_response, service2_response, service3_response):
    # Эта функция не делает сетевых вызовов, только обрабатывает ответы
    # Считайте, что она уже написана
    return {}

# Схема работы:
# 1. Получаем item_ids с каждого запроса к 'some-service/getItems'
# 2. Для каждого item_id отправляем независимые запросы к service_1_url, service_2_url, service_3_url
# 3. Обрабатываем результаты через business_logic
# 4. Собираем все результаты в итоговый список

def gather_data(requests: List[tuple]) -> List[dict]:
    # YOUR CODE HERE
    pass

Ответ

Обрабатывайте вход ограниченными партиями, а три независимых запроса для item запускайте конкурентно. Ограничивайте число HTTP-операций и не создавайте миллион задач сразу.

Разделите конвейер на получение item_id и обработку item. Используйте общий HTTP-клиент, ограниченную очередь и фиксированные workers. На один item запускайте три запроса, ждите все ответы и передавайте их в business_logic; каждому сервису задайте собственные лимиты, timeout и политику повторов.

Функция gather_data может быть синхронной обёрткой над async-конвейером, но asyncio.run нельзя вызывать внутри уже работающего event loop. Сохранение всех результатов в финальном списке неизбежно требует O(R) памяти. При слишком большом R контракт нужно изменить на поток или хранилище результатов. Зафиксируйте порядок, правила частичных ошибок, дедупликацию item_id и идемпотентность сервиса логирования.