import json import time import requests from fatcat_tools.workers.worker_common import FatcatWorker from fatcat_client import ReleaseEntity from fatcat_tools.transforms import * from pykafka.common import OffsetType class FatcatElasticReleaseWorker(FatcatWorker): """ Consumes from release-updates topic and pushes into (presumably local) elasticsearch. Uses a consumer group to manage offset. """ def __init__(self, kafka_hosts, consume_topic, poll_interval=10.0, offset=None, elastic_backend="http://localhost:9200", elastic_index="fatcat"): super().__init__(kafka_hosts=kafka_hosts, consume_topic=consume_topic, api_host_url=None) self.consumer_group = "elastic-updates" self.elastic_backend = elastic_backend self.elastic_index = elastic_index def run(self): consume_topic = self.kafka.topics[self.consume_topic] consumer = consume_topic.get_balanced_consumer( consumer_group=self.consumer_group, managed=True, ) for msg in consumer: json_str = msg.value.decode('utf-8') release = entity_from_json(json_str, ReleaseEntity) #print(release) elastic_endpoint = "{}/{}/release/{}".format( self.elastic_backend, self.elastic_index, release.ident) print("Updating document: {}".format(elastic_endpoint)) resp = requests.post(elastic_endpoint, json=release.to_elastic_dict()) assert resp.status_code in (200, 201) consumer.commit_offsets()