Где хранить информацию полученную в результате работы нескольких корутин?

Question

[Александр]

Пишу ассинхронный парсер, моя логика такая, собираю ссылки на продукты их где то 800 штук, дальше у меня в функции есть такой код:

product_links = await get_category_products_links(page_links)
print(product_links)

parser_tasks = []

for product_link in product_links:
     parser_task = asyncio.create_task(set_product_info(task_id, category['id'], product_link))
     parser_tasks.append(parser_task)
        
await asyncio.gather(*parser_tasks)

корутина set_products_info():
В ней только одна строка product_info = await get_product_info(product_link), подскажите удобный способ поместить куда либо информацию о продукте, чтобы потом взять всю информацию о продуктах и записать их в файл?
Тоесть мне нужно из каждого вызова положить все product_info в одно место.
Базу данных не могу использовать по заданию, думал в глобальную переменную, но какой она должна быть, не знаю (список, очередь)?
Пытался писать в файл с помощью библиотеки aiofile, но у меня ломается текст.

Answer 1

[Андрей Шубин]

Воспользуйтесь очередью - https://docs.python.org/3/library/asyncio-queue.html
Создайте её, прокиньте в таски, и сделайте корутину, которая будет вычитывать сообщения из очереди и складывать куда вам угодно, например, писать в файл.

Comments

[Александр]

Андрей Шубин , хорошо попробую, а вот такой вопрос у меня, то что у меня 800 тасков помещается в asyncio.gather() это нормально или это не правильно в плане логики?

[Андрей Шубин]

Александр, это нужно тестировать. Замерить сколько времени занимает прогнать 800 тасок сразу, или сделать так, чтобы одновременно работало только 20 воркеров - сколько это займет времени? А 40? А 10? И так далее. Я думаю это может зависеть от того насколько тяжелая таска. Буду рад, если кто-то меня поправит.

С воркерами может понадобиться две очереди, из одной воркеры будут получать следующую ссылку, в другую писать результат.

[javedimka]

Андрей Шубин, Что имеется ввиду под воркерами?

[Андрей Шубин]

javedimka, те же таски, которые вы запускаете в необходимом количестве. Далее они вычитывают из очереди задач сами задачи, выполняют их, и записывают результат в очередь результатов. Ну и читают до тех пор, пока в очереди есть задачи.

Накидал небольной пример. Очередь q содержит задачи (ссылки в вашем случае), воркеры читают из неё. Очередь r нужна для результатов, туда воркеры пишут, а logger читает и пишет все в лог.

Теоретически очередь можно тоже постепенно наполнять, например у вас файл с ссылками, считали из него 100 ссылок, закинули в очередь, очередь опустела, считали следующие 100 ссылок. Когда считывать нечего - прибиваете воркеры, завершаете скрипт.

import asyncio
import logging
import random

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')


async def logger(r: asyncio.Queue):
    while True:
        msg = await r.get()
        logging.info(msg)


async def worker(q: asyncio.Queue, r: asyncio.Queue, index: int):
    while not q.empty():
        task = await q.get()
        await r.put(f'Worker #{index} performs task #{task}')
        await asyncio.sleep(random.random())
    await r.put(f'Worker #{index} finished')


async def main(loop: asyncio.AbstractEventLoop):
    WORKERS_AMOUNT = 5
    q = asyncio.Queue()
    r = asyncio.Queue()
    tasks: list = []
    [await q.put(x) for x in range(100)]

    l = loop.create_task(logger(r))

    for i in range(WORKERS_AMOUNT):
        tasks.append(loop.create_task(worker(q, r, i)))
    await asyncio.gather(*tasks)
    l.cancel()

if __name__ == '__main__':
    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()

[javedimka]

Андрей Шубин, Я в асинкио бум-бум и понятие "воркер" у меня в голове = тред/процесс. Если не сложно, то можно поподробнее объяснить то в чем тут разница? Большая ли разница во времени выполнения скрипта будет между 5 корутинами которые ждут эвентов на сокетах и 800 корутинами? Если да, то откуда она появится?

[Андрей Шубин]

javedimka, как я написал выше, это нужно проверить. Я не могу ответить на этот вопрос. Если в вашем случае 800 корутин работает, то ок. Просто если со временем их станет больше, то как поведет себя скрипт?

Я в асинкио бум-бум и понятие "воркер" у меня в голове = тред/процесс.

Да воркер это и есть корутина и в принципе тот же тред, ибо треды тоже не параллельно выполняются.

[Александр]

Андрей Шубин, спасибо, я понял, но выше не я вам писал на счёт воркеров)

[Андрей Шубин]

Александр, да, я тупанул.