fmt (black): fatcat_tools/

3 files changed, 263 insertions, 196 deletions

diff --git a/python/fatcat_tools/workers/changelog.py b/python/fatcat_tools/workers/changelog.py
index a61e364c..1e4cb41d 100644
--- a/python/fatcat_tools/workers/changelog.py
+++ b/python/fatcat_tools/workers/changelog.py
@@ -1,4 +1,3 @@
-
 import json
 import time
 
@@ -16,11 +15,9 @@ class ChangelogWorker(FatcatWorker):
     """
 
     def __init__(self, api, kafka_hosts, produce_topic, poll_interval=10.0, offset=None):
-        super().__init__(kafka_hosts=kafka_hosts,
-                         produce_topic=produce_topic,
-                         api=api)
+        super().__init__(kafka_hosts=kafka_hosts, produce_topic=produce_topic, api=api)
         self.poll_interval = poll_interval
-        self.offset = offset    # the fatcat changelog offset, not the kafka offset
+        self.offset = offset  # the fatcat changelog offset, not the kafka offset
 
     def run(self):
 
@@ -31,7 +28,7 @@ class ChangelogWorker(FatcatWorker):
             print("Checking for most recent changelog offset...")
             msg = most_recent_message(self.produce_topic, self.kafka_config)
             if msg:
-                self.offset = json.loads(msg.decode('utf-8'))['index']
+                self.offset = json.loads(msg.decode("utf-8"))["index"]
             else:
                 self.offset = 0
             print("Most recent changelog index in Kafka seems to be {}".format(self.offset))
@@ -44,28 +41,29 @@ class ChangelogWorker(FatcatWorker):
                 raise KafkaException(err)
 
         producer_conf = self.kafka_config.copy()
-        producer_conf.update({
-            'delivery.report.only.error': True,
-            'default.topic.config': {
-                'request.required.acks': -1, # all brokers must confirm
-            },
-        })
+        producer_conf.update(
+            {
+                "delivery.report.only.error": True,
+                "default.topic.config": {
+                    "request.required.acks": -1,  # all brokers must confirm
+                },
+            }
+        )
         producer = Producer(producer_conf)
 
         while True:
             latest = int(self.api.get_changelog(limit=1)[0].index)
             if latest > self.offset:
-                print("Fetching changelogs from {} through {}".format(
-                    self.offset+1, latest))
-            for i in range(self.offset+1, latest+1):
+                print("Fetching changelogs from {} through {}".format(self.offset + 1, latest))
+            for i in range(self.offset + 1, latest + 1):
                 cle = self.api.get_changelog_entry(i)
                 obj = self.api.api_client.sanitize_for_serialization(cle)
                 producer.produce(
                     self.produce_topic,
-                    json.dumps(obj).encode('utf-8'),
+                    json.dumps(obj).encode("utf-8"),
                     key=str(i),
                     on_delivery=fail_fast,
-                    #NOTE timestamp could be timestamp=cle.timestamp (?)
+                    # NOTE timestamp could be timestamp=cle.timestamp (?)
                 )
                 self.offset = i
             producer.flush()
@@ -79,12 +77,19 @@ class EntityUpdatesWorker(FatcatWorker):
     from API) to update topics.
     """
 
-    def __init__(self, api, kafka_hosts, consume_topic, release_topic,
-            file_topic, container_topic, ingest_file_request_topic,
-            work_ident_topic, poll_interval=5.0):
-        super().__init__(kafka_hosts=kafka_hosts,
-                         consume_topic=consume_topic,
-                         api=api)
+    def __init__(
+        self,
+        api,
+        kafka_hosts,
+        consume_topic,
+        release_topic,
+        file_topic,
+        container_topic,
+        ingest_file_request_topic,
+        work_ident_topic,
+        poll_interval=5.0,
+    ):
+        super().__init__(kafka_hosts=kafka_hosts, consume_topic=consume_topic, api=api)
         self.release_topic = release_topic
         self.file_topic = file_topic
         self.container_topic = container_topic
@@ -150,7 +155,7 @@ class EntityUpdatesWorker(FatcatWorker):
             # Transactions of the Japan Society of Mechanical Engineers
             "10.1299/kikai",
             # protocols.io
-            "10.17504/"
+            "10.17504/",
         ]
 
     def want_live_ingest(self, release, ingest_request):
@@ -163,40 +168,40 @@ class EntityUpdatesWorker(FatcatWorker):
         ingest crawling (via wayback SPN).
         """
 
-        link_source = ingest_request.get('ingest_request')
-        ingest_type = ingest_request.get('ingest_type')
-        doi = ingest_request.get('ext_ids', {}).get('doi')
+        link_source = ingest_request.get("ingest_request")
+        ingest_type = ingest_request.get("ingest_type")
+        doi = ingest_request.get("ext_ids", {}).get("doi")
         es = release_to_elasticsearch(release)
 
         is_document = release.release_type in (
-            'article',
-            'article-journal',
-            'article-newspaper',
-            'book',
-            'chapter',
-            'editorial',
-            'interview',
-            'legal_case',
-            'legislation',
-            'letter',
-            'manuscript',
-            'paper-conference',
-            'patent',
-            'peer_review',
-            'post',
-            'report',
-            'retraction',
-            'review',
-            'review-book',
-            'thesis',
+            "article",
+            "article-journal",
+            "article-newspaper",
+            "book",
+            "chapter",
+            "editorial",
+            "interview",
+            "legal_case",
+            "legislation",
+            "letter",
+            "manuscript",
+            "paper-conference",
+            "patent",
+            "peer_review",
+            "post",
+            "report",
+            "retraction",
+            "review",
+            "review-book",
+            "thesis",
         )
         is_not_pdf = release.release_type in (
-            'component',
-            'dataset',
-            'figure',
-            'graphic',
-            'software',
-            'stub',
+            "component",
+            "dataset",
+            "figure",
+            "graphic",
+            "software",
+            "stub",
         )
 
         # accept list sets a default "crawl it" despite OA metadata for
@@ -207,19 +212,23 @@ class EntityUpdatesWorker(FatcatWorker):
                 if doi.startswith(prefix):
                     in_acceptlist = True
 
-        if self.ingest_oa_only and link_source not in ('arxiv', 'pmc'):
+        if self.ingest_oa_only and link_source not in ("arxiv", "pmc"):
 
             # most datacite documents are in IRs and should be crawled
             is_datacite_doc = False
-            if release.extra and ('datacite' in release.extra) and is_document:
+            if release.extra and ("datacite" in release.extra) and is_document:
                 is_datacite_doc = True
-            if not (es['is_oa'] or in_acceptlist or is_datacite_doc):
+            if not (es["is_oa"] or in_acceptlist or is_datacite_doc):
                 return False
 
         # big publishers *generally* have accurate OA metadata, use
         # preservation networks, and block our crawlers. So unless OA, or
         # explicitly on accept list, or not preserved, skip crawling
-        if es.get('publisher_type') == 'big5' and es.get('is_preserved') and not (es['is_oa'] or in_acceptlist):
+        if (
+            es.get("publisher_type") == "big5"
+            and es.get("is_preserved")
+            and not (es["is_oa"] or in_acceptlist)
+        ):
             return False
 
         # if ingest_type is pdf but release_type is almost certainly not a PDF,
@@ -233,23 +242,24 @@ class EntityUpdatesWorker(FatcatWorker):
                     return False
 
         # figshare
-        if doi and (doi.startswith('10.6084/') or doi.startswith('10.25384/')):
+        if doi and (doi.startswith("10.6084/") or doi.startswith("10.25384/")):
             # don't crawl "most recent version" (aka "group") DOIs
             if not release.version:
                 return False
 
         # zenodo
-        if doi and doi.startswith('10.5281/'):
+        if doi and doi.startswith("10.5281/"):
             # if this is a "grouping" DOI of multiple "version" DOIs, do not crawl (will crawl the versioned DOIs)
-            if release.extra and release.extra.get('relations'):
-                for rel in release.extra['relations']:
-                    if (rel.get('relationType') == 'HasVersion' and rel.get('relatedIdentifier', '').startswith('10.5281/')):
+            if release.extra and release.extra.get("relations"):
+                for rel in release.extra["relations"]:
+                    if rel.get("relationType") == "HasVersion" and rel.get(
+                        "relatedIdentifier", ""
+                    ).startswith("10.5281/"):
                         return False
 
         return True
 
     def run(self):
-
         def fail_fast(err, msg):
             if err is not None:
                 print("Kafka producer delivery error: {}".format(err))
@@ -278,36 +288,40 @@ class EntityUpdatesWorker(FatcatWorker):
             for p in partitions:
                 if p.error:
                     raise KafkaException(p.error)
-            print("Kafka partitions rebalanced: {} / {}".format(
-                consumer, partitions))
+            print("Kafka partitions rebalanced: {} / {}".format(consumer, partitions))
 
         consumer_conf = self.kafka_config.copy()
-        consumer_conf.update({
-            'group.id': self.consumer_group,
-            'on_commit': fail_fast,
-            # messages don't have offset marked as stored until pushed to
-            # elastic, but we do auto-commit stored offsets to broker
-            'enable.auto.commit': True,
-            'enable.auto.offset.store': False,
-            # user code timeout; if no poll after this long, assume user code
-            # hung and rebalance (default: 5min)
-            'max.poll.interval.ms': 180000,
-            'default.topic.config': {
-                'auto.offset.reset': 'latest',
-            },
-        })
+        consumer_conf.update(
+            {
+                "group.id": self.consumer_group,
+                "on_commit": fail_fast,
+                # messages don't have offset marked as stored until pushed to
+                # elastic, but we do auto-commit stored offsets to broker
+                "enable.auto.commit": True,
+                "enable.auto.offset.store": False,
+                # user code timeout; if no poll after this long, assume user code
+                # hung and rebalance (default: 5min)
+                "max.poll.interval.ms": 180000,
+                "default.topic.config": {
+                    "auto.offset.reset": "latest",
+                },
+            }
+        )
         consumer = Consumer(consumer_conf)
 
         producer_conf = self.kafka_config.copy()
-        producer_conf.update({
-            'delivery.report.only.error': True,
-            'default.topic.config': {
-                'request.required.acks': -1, # all brokers must confirm
-            },
-        })
+        producer_conf.update(
+            {
+                "delivery.report.only.error": True,
+                "default.topic.config": {
+                    "request.required.acks": -1,  # all brokers must confirm
+                },
+            }
+        )
         producer = Producer(producer_conf)
 
-        consumer.subscribe([self.consume_topic],
+        consumer.subscribe(
+            [self.consume_topic],
             on_assign=on_rebalance,
             on_revoke=on_rebalance,
         )
@@ -316,14 +330,16 @@ class EntityUpdatesWorker(FatcatWorker):
         while True:
             msg = consumer.poll(self.poll_interval)
             if not msg:
-                print("nothing new from kafka (poll_interval: {} sec)".format(self.poll_interval))
+                print(
+                    "nothing new from kafka (poll_interval: {} sec)".format(self.poll_interval)
+                )
                 continue
             if msg.error():
                 raise KafkaException(msg.error())
 
-            cle = json.loads(msg.value().decode('utf-8'))
-            #print(cle)
-            print("processing changelog index {}".format(cle['index']))
+            cle = json.loads(msg.value().decode("utf-8"))
+            # print(cle)
+            print("processing changelog index {}".format(cle["index"]))
             release_ids = []
             new_release_ids = []
             file_ids = []
@@ -331,27 +347,27 @@ class EntityUpdatesWorker(FatcatWorker):
             webcapture_ids = []
             container_ids = []
             work_ids = []
-            release_edits = cle['editgroup']['edits']['releases']
+            release_edits = cle["editgroup"]["edits"]["releases"]
             for re in release_edits:
-                release_ids.append(re['ident'])
+                release_ids.append(re["ident"])
                 # filter to direct release edits which are not updates
-                if not re.get('prev_revision') and not re.get('redirect_ident'):
-                    new_release_ids.append(re['ident'])
-            file_edits = cle['editgroup']['edits']['files']
+                if not re.get("prev_revision") and not re.get("redirect_ident"):
+                    new_release_ids.append(re["ident"])
+            file_edits = cle["editgroup"]["edits"]["files"]
             for e in file_edits:
-                file_ids.append(e['ident'])
-            fileset_edits = cle['editgroup']['edits']['filesets']
+                file_ids.append(e["ident"])
+            fileset_edits = cle["editgroup"]["edits"]["filesets"]
             for e in fileset_edits:
-                fileset_ids.append(e['ident'])
-            webcapture_edits = cle['editgroup']['edits']['webcaptures']
+                fileset_ids.append(e["ident"])
+            webcapture_edits = cle["editgroup"]["edits"]["webcaptures"]
             for e in webcapture_edits:
-                webcapture_ids.append(e['ident'])
-            container_edits = cle['editgroup']['edits']['containers']
+                webcapture_ids.append(e["ident"])
+            container_edits = cle["editgroup"]["edits"]["containers"]
             for e in container_edits:
-                container_ids.append(e['ident'])
-            work_edits = cle['editgroup']['edits']['works']
+                container_ids.append(e["ident"])
+            work_edits = cle["editgroup"]["edits"]["works"]
             for e in work_edits:
-                work_ids.append(e['ident'])
+                work_ids.append(e["ident"])
 
             # TODO: do these fetches in parallel using a thread pool?
             for ident in set(file_ids):
@@ -363,8 +379,8 @@ class EntityUpdatesWorker(FatcatWorker):
                 file_dict = self.api.api_client.sanitize_for_serialization(file_entity)
                 producer.produce(
                     self.file_topic,
-                    json.dumps(file_dict).encode('utf-8'),
-                    key=ident.encode('utf-8'),
+                    json.dumps(file_dict).encode("utf-8"),
+                    key=ident.encode("utf-8"),
                     on_delivery=fail_fast,
                 )
 
@@ -385,30 +401,34 @@ class EntityUpdatesWorker(FatcatWorker):
                 container_dict = self.api.api_client.sanitize_for_serialization(container)
                 producer.produce(
                     self.container_topic,
-                    json.dumps(container_dict).encode('utf-8'),
-                    key=ident.encode('utf-8'),
+                    json.dumps(container_dict).encode("utf-8"),
+                    key=ident.encode("utf-8"),
                     on_delivery=fail_fast,
                 )
 
             for ident in set(release_ids):
-                release = self.api.get_release(ident, expand="files,filesets,webcaptures,container")
+                release = self.api.get_release(
+                    ident, expand="files,filesets,webcaptures,container"
+                )
                 if release.work_id:
                     work_ids.append(release.work_id)
                 release_dict = self.api.api_client.sanitize_for_serialization(release)
                 producer.produce(
                     self.release_topic,
-                    json.dumps(release_dict).encode('utf-8'),
-                    key=ident.encode('utf-8'),
+                    json.dumps(release_dict).encode("utf-8"),
+                    key=ident.encode("utf-8"),
                     on_delivery=fail_fast,
                 )
                 # for ingest requests, filter to "new" active releases with no matched files
                 if release.ident in new_release_ids:
-                    ir = release_ingest_request(release, ingest_request_source='fatcat-changelog')
+                    ir = release_ingest_request(
+                        release, ingest_request_source="fatcat-changelog"
+                    )
                     if ir and not release.files and self.want_live_ingest(release, ir):
                         producer.produce(
                             self.ingest_file_request_topic,
-                            json.dumps(ir).encode('utf-8'),
-                            #key=None,
+                            json.dumps(ir).encode("utf-8"),
+                            # key=None,
                             on_delivery=fail_fast,
                         )
 
@@ -420,13 +440,13 @@ class EntityUpdatesWorker(FatcatWorker):
                     key=key,
                     type="fatcat_work",
                     work_ident=ident,
-                    updated=cle['timestamp'],
-                    fatcat_changelog_index=cle['index'],
+                    updated=cle["timestamp"],
+                    fatcat_changelog_index=cle["index"],
                 )
                 producer.produce(
                     self.work_ident_topic,
-                    json.dumps(work_ident_dict).encode('utf-8'),
-                    key=key.encode('utf-8'),
+                    json.dumps(work_ident_dict).encode("utf-8"),
+                    key=key.encode("utf-8"),
                     on_delivery=fail_fast,
                 )
 
diff --git a/python/fatcat_tools/workers/elasticsearch.py b/python/fatcat_tools/workers/elasticsearch.py
index f411073d..0d75f964 100644
--- a/python/fatcat_tools/workers/elasticsearch.py
+++ b/python/fatcat_tools/workers/elasticsearch.py
@@ -1,4 +1,3 @@
-
 import json
 import sys
 
@@ -26,12 +25,20 @@ class ElasticsearchReleaseWorker(FatcatWorker):
     Uses a consumer group to manage offset.
     """
 
-    def __init__(self, kafka_hosts, consume_topic, poll_interval=10.0, offset=None,
-            elasticsearch_backend="http://localhost:9200", elasticsearch_index="fatcat",
-            elasticsearch_release_index="fatcat_releases",
-            batch_size=200, api_host="https://api.fatcat.wiki/v0", query_stats=False):
-        super().__init__(kafka_hosts=kafka_hosts,
-                         consume_topic=consume_topic)
+    def __init__(
+        self,
+        kafka_hosts,
+        consume_topic,
+        poll_interval=10.0,
+        offset=None,
+        elasticsearch_backend="http://localhost:9200",
+        elasticsearch_index="fatcat",
+        elasticsearch_release_index="fatcat_releases",
+        batch_size=200,
+        api_host="https://api.fatcat.wiki/v0",
+        query_stats=False,
+    ):
+        super().__init__(kafka_hosts=kafka_hosts, consume_topic=consume_topic)
         self.consumer_group = "elasticsearch-updates3"
         self.batch_size = batch_size
         self.poll_interval = poll_interval
@@ -63,45 +70,53 @@ class ElasticsearchReleaseWorker(FatcatWorker):
                     print("Bailing out...", file=sys.stderr)
                     # TODO: should it be sys.exit(-1)?
                     raise KafkaException(p.error)
-            #print("Kafka consumer commit successful")
+            # print("Kafka consumer commit successful")
             pass
 
         def on_rebalance(consumer, partitions):
             for p in partitions:
                 if p.error:
                     raise KafkaException(p.error)
-            print("Kafka partitions rebalanced: {} / {}".format(
-                consumer, partitions), file=sys.stderr)
+            print(
+                "Kafka partitions rebalanced: {} / {}".format(consumer, partitions),
+                file=sys.stderr,
+            )
 
         consumer_conf = self.kafka_config.copy()
-        consumer_conf.update({
-            'group.id': self.consumer_group,
-            'on_commit': fail_fast,
-            # messages don't have offset marked as stored until pushed to
-            # elastic, but we do auto-commit stored offsets to broker
-            'enable.auto.commit': True,
-            'enable.auto.offset.store': False,
-            # user code timeout; if no poll after this long, assume user code
-            # hung and rebalance (default: 5min)
-            'max.poll.interval.ms': 60000,
-            'default.topic.config': {
-                'auto.offset.reset': 'latest',
-            },
-        })
+        consumer_conf.update(
+            {
+                "group.id": self.consumer_group,
+                "on_commit": fail_fast,
+                # messages don't have offset marked as stored until pushed to
+                # elastic, but we do auto-commit stored offsets to broker
+                "enable.auto.commit": True,
+                "enable.auto.offset.store": False,
+                # user code timeout; if no poll after this long, assume user code
+                # hung and rebalance (default: 5min)
+                "max.poll.interval.ms": 60000,
+                "default.topic.config": {
+                    "auto.offset.reset": "latest",
+                },
+            }
+        )
         consumer = Consumer(consumer_conf)
-        consumer.subscribe([self.consume_topic],
+        consumer.subscribe(
+            [self.consume_topic],
             on_assign=on_rebalance,
             on_revoke=on_rebalance,
         )
 
         while True:
-            batch = consumer.consume(
-                num_messages=self.batch_size,
-                timeout=self.poll_interval)
+            batch = consumer.consume(num_messages=self.batch_size, timeout=self.poll_interval)
             if not batch:
                 if not consumer.assignment():
                     print("... no Kafka consumer partitions assigned yet", file=sys.stderr)
-                print("... nothing new from kafka, try again (interval: {}".format(self.poll_interval), file=sys.stderr)
+                print(
+                    "... nothing new from kafka, try again (interval: {}".format(
+                        self.poll_interval
+                    ),
+                    file=sys.stderr,
+                )
                 continue
             print("... got {} kafka messages".format(len(batch)), file=sys.stderr)
             # first check errors on entire batch...
@@ -111,19 +126,24 @@ class ElasticsearchReleaseWorker(FatcatWorker):
             # ... then process
             bulk_actions = []
             for msg in batch:
-                json_str = msg.value().decode('utf-8')
+                json_str = msg.value().decode("utf-8")
                 entity = entity_from_json(json_str, self.entity_type, api_client=ac)
                 assert isinstance(entity, self.entity_type)
                 if self.entity_type == ChangelogEntry:
                     key = entity.index
                     # might need to fetch from API
-                    if not (entity.editgroup and entity.editgroup.editor): # pylint: disable=no-member # (TODO)
+                    if not (
+                        entity.editgroup and entity.editgroup.editor
+                    ):  # pylint: disable=no-member # (TODO)
                         entity = api.get_changelog_entry(entity.index)
                 else:
                     key = entity.ident  # pylint: disable=no-member # (TODO)
 
-                if self.entity_type != ChangelogEntry and entity.state == 'wip':
-                    print(f"WARNING: skipping state=wip entity: {self.entity_type.__name__} {entity.ident}", file=sys.stderr)
+                if self.entity_type != ChangelogEntry and entity.state == "wip":
+                    print(
+                        f"WARNING: skipping state=wip entity: {self.entity_type.__name__} {entity.ident}",
+                        file=sys.stderr,
+                    )
                     continue
 
                 if self.entity_type == ContainerEntity and self.query_stats:
@@ -138,9 +158,15 @@ class ElasticsearchReleaseWorker(FatcatWorker):
                     doc_dict = self.transform_func(entity)
 
                 # TODO: handle deletions from index
-                bulk_actions.append(json.dumps({
-                    "index": { "_id": key, },
-                }))
+                bulk_actions.append(
+                    json.dumps(
+                        {
+                            "index": {
+                                "_id": key,
+                            },
+                        }
+                    )
+                )
                 bulk_actions.append(json.dumps(doc_dict))
 
             # if only WIP entities, then skip
@@ -149,15 +175,22 @@ class ElasticsearchReleaseWorker(FatcatWorker):
                     consumer.store_offsets(message=msg)
                 continue
 
-            print("Upserting, eg, {} (of {} {} in elasticsearch)".format(key, len(batch), self.entity_type.__name__), file=sys.stderr)
+            print(
+                "Upserting, eg, {} (of {} {} in elasticsearch)".format(
+                    key, len(batch), self.entity_type.__name__
+                ),
+                file=sys.stderr,
+            )
             elasticsearch_endpoint = "{}/{}/_bulk".format(
-                self.elasticsearch_backend,
-                self.elasticsearch_index)
-            resp = requests.post(elasticsearch_endpoint,
+                self.elasticsearch_backend, self.elasticsearch_index
+            )
+            resp = requests.post(
+                elasticsearch_endpoint,
                 headers={"Content-Type": "application/x-ndjson"},
-                data="\n".join(bulk_actions) + "\n")
+                data="\n".join(bulk_actions) + "\n",
+            )
             resp.raise_for_status()
-            if resp.json()['errors']:
+            if resp.json()["errors"]:
                 desc = "Elasticsearch errors from post to {}:".format(elasticsearch_endpoint)
                 print(desc, file=sys.stderr)
                 print(resp.content, file=sys.stderr)
@@ -169,20 +202,29 @@ class ElasticsearchReleaseWorker(FatcatWorker):
 
 
 class ElasticsearchContainerWorker(ElasticsearchReleaseWorker):
-
-    def __init__(self, kafka_hosts, consume_topic, poll_interval=10.0, offset=None,
-            query_stats=False, elasticsearch_release_index="fatcat_release",
-            elasticsearch_backend="http://localhost:9200", elasticsearch_index="fatcat",
-            batch_size=200):
-        super().__init__(kafka_hosts=kafka_hosts,
-                         consume_topic=consume_topic,
-                         poll_interval=poll_interval,
-                         offset=offset,
-                         elasticsearch_backend=elasticsearch_backend,
-                         elasticsearch_index=elasticsearch_index,
-                         elasticsearch_release_index=elasticsearch_release_index,
-                         query_stats=query_stats,
-                         batch_size=batch_size)
+    def __init__(
+        self,
+        kafka_hosts,
+        consume_topic,
+        poll_interval=10.0,
+        offset=None,
+        query_stats=False,
+        elasticsearch_release_index="fatcat_release",
+        elasticsearch_backend="http://localhost:9200",
+        elasticsearch_index="fatcat",
+        batch_size=200,
+    ):
+        super().__init__(
+            kafka_hosts=kafka_hosts,
+            consume_topic=consume_topic,
+            poll_interval=poll_interval,
+            offset=offset,
+            elasticsearch_backend=elasticsearch_backend,
+            elasticsearch_index=elasticsearch_index,
+            elasticsearch_release_index=elasticsearch_release_index,
+            query_stats=query_stats,
+            batch_size=batch_size,
+        )
         # previous group got corrupted (by pykafka library?)
         self.consumer_group = "elasticsearch-updates3"
         self.entity_type = ContainerEntity
@@ -196,11 +238,18 @@ class ElasticsearchChangelogWorker(ElasticsearchReleaseWorker):
     Note: Very early versions of changelog entries did not contain details
     about the editor or extra fields.
     """
-    def __init__(self, kafka_hosts, consume_topic, poll_interval=10.0, offset=None,
-            elasticsearch_backend="http://localhost:9200", elasticsearch_index="fatcat_changelog",
-            batch_size=200):
-        super().__init__(kafka_hosts=kafka_hosts,
-                         consume_topic=consume_topic)
+
+    def __init__(
+        self,
+        kafka_hosts,
+        consume_topic,
+        poll_interval=10.0,
+        offset=None,
+        elasticsearch_backend="http://localhost:9200",
+        elasticsearch_index="fatcat_changelog",
+        batch_size=200,
+    ):
+        super().__init__(kafka_hosts=kafka_hosts, consume_topic=consume_topic)
         self.consumer_group = "elasticsearch-updates3"
         self.batch_size = batch_size
         self.poll_interval = poll_interval
diff --git a/python/fatcat_tools/workers/worker_common.py b/python/fatcat_tools/workers/worker_common.py
index 8c2936be..baec44f4 100644
--- a/python/fatcat_tools/workers/worker_common.py
+++ b/python/fatcat_tools/workers/worker_common.py
@@ -1,4 +1,3 @@
-
 from confluent_kafka import Consumer, KafkaException, TopicPartition
 
 
@@ -13,22 +12,21 @@ def most_recent_message(topic, kafka_config):
     print("Fetching most Kafka message from {}".format(topic))
 
     conf = kafka_config.copy()
-    conf.update({
-        'group.id': 'worker-init-last-msg', # should never commit
-        'delivery.report.only.error': True,
-        'enable.auto.commit': False,
-        'default.topic.config': {
-            'request.required.acks': -1,
-            'auto.offset.reset': 'latest',
-        },
-    })
+    conf.update(
+        {
+            "group.id": "worker-init-last-msg",  # should never commit
+            "delivery.report.only.error": True,
+            "enable.auto.commit": False,
+            "default.topic.config": {
+                "request.required.acks": -1,
+                "auto.offset.reset": "latest",
+            },
+        }
+    )
 
     consumer = Consumer(conf)
 
-    hwm = consumer.get_watermark_offsets(
-        TopicPartition(topic, 0),
-        timeout=5.0,
-        cached=False)
+    hwm = consumer.get_watermark_offsets(TopicPartition(topic, 0), timeout=5.0, cached=False)
     if not hwm:
         raise Exception("Kafka consumer timeout, or topic {} doesn't exist".format(topic))
     print("High watermarks: {}".format(hwm))
@@ -37,7 +35,7 @@ def most_recent_message(topic, kafka_config):
         print("topic is new; not 'most recent message'")
         return None
 
-    consumer.assign([TopicPartition(topic, 0, hwm[1]-1)])
+    consumer.assign([TopicPartition(topic, 0, hwm[1] - 1)])
     msg = consumer.poll(2.0)
     consumer.close()
     if not msg:
@@ -56,8 +54,8 @@ class FatcatWorker:
         if api:
             self.api = api
         self.kafka_config = {
-            'bootstrap.servers': kafka_hosts,
-            'message.max.bytes': 20000000, # ~20 MBytes; broker-side max is ~50 MBytes
+            "bootstrap.servers": kafka_hosts,
+            "message.max.bytes": 20000000,  # ~20 MBytes; broker-side max is ~50 MBytes
         }
         self.produce_topic = produce_topic
         self.consume_topic = consume_topic