Clean up old linter complaints

Author: cnweaver

adc/consumer.py

@@ -38,7 +38,8 @@ def __init__(self, conf: 'ConsumerConfig') -> None:
         self.logger = logging.getLogger("adc-streaming.consumer")
         self.conf = conf
         self._consumer = confluent_kafka.Consumer(conf._to_confluent_kafka())
-        # Workaround for https://github.com/confluentinc/librdkafka/issues/3753#issuecomment-1058272987.
+        # Workaround for
+        # https://github.com/confluentinc/librdkafka/issues/3753#issuecomment-1058272987.
         # FIXME: Remove once fixed upstream, or on removal of oauth_cb.
         self._consumer.poll(0)
         self._stop_event = threading.Event()
@@ -196,7 +197,7 @@ def _stream_forever(self,
                             self.mark_done(m, asynchronous=True)
                         yield m
                     else:
-                        raise(confluent_kafka.KafkaException(err))
+                        raise (confluent_kafka.KafkaException(err))
             finally:
                 if autocommit:
                     self._consumer.commit(asynchronous=True)
@@ -242,7 +243,7 @@ def _stream_until_eof(self,
                             # Done with all partitions for the topic, remove it
                             del active_partitions[m.topic()]
                     else:
-                        raise(confluent_kafka.KafkaException(err))
+                        raise (confluent_kafka.KafkaException(err))
             finally:
                 if autocommit:
                     self._consumer.commit(asynchronous=True)

adc/producer.py

@@ -27,7 +27,8 @@ def __init__(self, conf: 'ProducerConfig') -> None:
         self.conf = conf
         self.logger.debug(f"connecting to producer with config {conf._to_confluent_kafka()}")
         self._producer = confluent_kafka.Producer(conf._to_confluent_kafka())
-        # Workaround for https://github.com/confluentinc/librdkafka/issues/3753#issuecomment-1058272987.
+        # Workaround for
+        # https://github.com/confluentinc/librdkafka/issues/3753#issuecomment-1058272987.
         # FIXME: Remove once fixed upstream, or on removal of oauth_cb.
         self._producer.poll(0)
 

tests/test_auth.py

@@ -1,6 +1,6 @@
 import pytest
 
-from adc.auth import SASLAuth, SASLMethod
+from adc.auth import SASLAuth
 
 
 @pytest.mark.parametrize('auth,expected_config', [

tests/test_kafka_integration.py

@@ -2,7 +2,7 @@
 import tempfile
 import time
 import unittest
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timedelta
 from typing import List
 
 import docker
@@ -283,7 +283,7 @@ def test_consume_from_datetime(self):
             "message 1",
             "message 2",
             "message 3",
-        ])        
+        ])
         # Wait a while, write, and wait some more
         time.sleep(2)
         client_middle_time = datetime.now()
@@ -390,8 +390,8 @@ def test_multi_topic_handling(self):
             topic=None,
             auth=self.kafka.auth,
         ))
-        for i in range(0,8):
-            producer.write(str(i), topic=topics[i%2])
+        for i in range(0, 8):
+            producer.write(str(i), topic=topics[i % 2])
         producer.flush()
         logger.info("messages sent")
 
@@ -403,12 +403,12 @@ def test_multi_topic_handling(self):
         ))
         consumer.subscribe(topics)
         stream = consumer.stream()
-        total_messages = 0;
+        total_messages = 0
         for msg in stream:
             if msg.error() is not None:
                 raise Exception(msg.error())
             idx = int(msg.value())
-            self.assertEqual(msg.topic(), topics[idx%2])
+            self.assertEqual(msg.topic(), topics[idx % 2])
             total_messages += 1
             if total_messages == 8:
                 break
@@ -554,7 +554,7 @@ def get_or_create_container(self):
             # these tests cannot run if there is already an instance of Kafka running on
             # the same host.
             ports={"9092/tcp": 9092},
-            command=["/root/runServer","--advertisedListener","SASL_SSL://localhost:9092"],
+            command=["/root/runServer", "--advertisedListener", "SASL_SSL://localhost:9092"],
         )
 
     def get_or_create_docker_network(self):