asyncworker_consumer.py

from typing import List, Dict, Iterable, Generator
from aioelasticsearch import Elasticsearch
from asyncworker import App, Options
from asyncworker.rabbitmq import RabbitMQMessage


app = App(host="localhost", user="guest", password="guest", prefetch_count=512)
es = Elasticsearch()


def gen_bulk(docs: Iterable[Dict]) -> Generator[Dict, None, None]:
    for doc in docs:
        yield {"index": {"_index": 'words', "_type": "doc"}}
        yield doc


@app.route(["words_to_index"], vhost="/", options={Options.BULK_SIZE: 256})
async def drain_handler(messages: List[RabbitMQMessage]):
    await es.bulk(body=gen_bulk((msg.body for msg in messages)))


app.run()