diff options
| author | Bryan Newbold <bnewbold@robocracy.org> | 2019-05-30 13:05:22 -0700 |
|---|---|---|
| committer | Bryan Newbold <bnewbold@robocracy.org> | 2019-05-30 13:05:22 -0700 |
| commit | 36bfdda186151972167c35cd0f642bc0c2ca1a54 (patch) | |
| tree | d1de856018885c9460739d2e2c3e120541adf12b | |
| parent | 393172a29e91bf1236e546b4a99f445dbbe849ef (diff) | |
| download | fatcat-36bfdda186151972167c35cd0f642bc0c2ca1a54.tar.gz fatcat-36bfdda186151972167c35cd0f642bc0c2ca1a54.zip | |
fix and workaround container entities in release topic
| -rw-r--r-- | python/fatcat_tools/workers/changelog.py | 4 | ||||
| -rw-r--r-- | python/fatcat_tools/workers/elasticsearch.py | 6 |
2 files changed, 8 insertions, 2 deletions
diff --git a/python/fatcat_tools/workers/changelog.py b/python/fatcat_tools/workers/changelog.py index 5ef1c69c..a753bb49 100644 --- a/python/fatcat_tools/workers/changelog.py +++ b/python/fatcat_tools/workers/changelog.py @@ -75,8 +75,8 @@ class EntityUpdatesWorker(FatcatWorker): def run(self): changelog_topic = self.kafka.topics[self.consume_topic] release_topic = self.kafka.topics[self.release_topic] - file_topic = self.kafka.topics[self.release_topic] - container_topic = self.kafka.topics[self.release_topic] + file_topic = self.kafka.topics[self.file_topic] + container_topic = self.kafka.topics[self.container_topic] consumer = changelog_topic.get_balanced_consumer( consumer_group=self.consumer_group, diff --git a/python/fatcat_tools/workers/elasticsearch.py b/python/fatcat_tools/workers/elasticsearch.py index fc96e3d4..a18ed244 100644 --- a/python/fatcat_tools/workers/elasticsearch.py +++ b/python/fatcat_tools/workers/elasticsearch.py @@ -43,6 +43,12 @@ class ElasticsearchReleaseWorker(FatcatWorker): for msg in consumer: json_str = msg.value.decode('utf-8') + # HACK: work around a bug where container entities got published to + # release_v03 topic + if self.elasticsearch_document_name == "release": + entity_dict = json.loads(json_str) + if entity_dict.get('name') and not entity_dict.get('title'): + continue entity = entity_from_json(json_str, self.entity_type, api_client=ac) #print(entity) elasticsearch_endpoint = "{}/{}/{}/{}".format( |