asyncworker_producer.py

import asyncio
from http import HTTPStatus
from typing import AsyncIterator

from aiohttp import ClientSession
from easyqueue import AsyncQueue

WORDLIST_URL = "https://s3.amazonaws.com/diogo.martins/public/portuguese-brazil.txt"
loop = asyncio.get_event_loop()


async def stream_wordlist(limit: int=None) -> AsyncIterator[str]:
    async with ClientSession() as session:
        async with session.get(WORDLIST_URL) as response:
            if response.status == HTTPStatus.OK:
                i = 0
                async for row in response.content:
                    yield row.rstrip().decode('utf-8')
                    i += 1
                    if i == limit:
                        break


async def produce_messages():
    amqp = AsyncQueue(host='localhost', username='guest', password='guest')
    await amqp.connect()

    tasks = []
    async for word in stream_wordlist(limit=10000):
        task = loop.create_task(
            amqp.put({'word': word}, routing_key='words', exchange='default')
        )
        tasks.append(task)

    await asyncio.gather(*tasks)

loop.run_until_complete(produce_messages())