ruff

Author: rerpha

src/saluki/consume.py

@@ -1,6 +1,7 @@
 import logging
 
 from confluent_kafka import Consumer, TopicPartition
+
 from saluki.utils import _deserialise_and_print_messages
 
 logger = logging.getLogger("saluki")
@@ -46,9 +47,7 @@ def consume(
             start = offset - num_messages + 1
         else:
             start = (
-                c.get_watermark_offsets(TopicPartition(topic, partition), cached=False)[
-                    1
-                ]
+                c.get_watermark_offsets(TopicPartition(topic, partition), cached=False)[1]
                 - num_messages
             )
 

src/saluki/listen.py

@@ -1,6 +1,7 @@
 import logging
 
 from confluent_kafka import Consumer, TopicPartition
+
 from saluki.utils import _deserialise_and_print_messages
 
 logger = logging.getLogger("saluki")

src/saluki/main.py

@@ -21,9 +21,7 @@ def main() -> None:
     )
 
     parent_parser = argparse.ArgumentParser(add_help=False)
-    parent_parser.add_argument(
-        "topic", type=str, help="Kafka topic. format is broker<:port>/topic"
-    )
+    parent_parser.add_argument("topic", type=str, help="Kafka topic. format is broker<:port>/topic")
 
     parent_parser.add_argument(
         "-X",
@@ -41,9 +39,7 @@ def main() -> None:
         type=argparse.FileType("a"),
     )
 
-    sub_parsers = parser.add_subparsers(
-        help="sub-command help", required=True, dest="command"
-    )
+    sub_parsers = parser.add_subparsers(help="sub-command help", required=True, dest="command")
 
     consumer_parser = argparse.ArgumentParser(add_help=False)
     consumer_parser.add_argument(
@@ -68,24 +64,16 @@ def main() -> None:
     consumer_mode_parser.add_argument(
         "-o", "--offset", help="offset to consume from", type=int, required=False
     )
-    consumer_mode_parser.add_argument(
-        "-s", "--schema", required=False, default="auto", type=str
-    )
-    consumer_mode_parser.add_argument(
-        "-g", "--go-forwards", required=False, action="store_true"
-    )
-    consumer_mode_parser.add_argument(
-        "-p", "--partition", required=False, type=int, default=0
-    )
+    consumer_mode_parser.add_argument("-s", "--schema", required=False, default="auto", type=str)
+    consumer_mode_parser.add_argument("-g", "--go-forwards", required=False, action="store_true")
+    consumer_mode_parser.add_argument("-p", "--partition", required=False, type=int, default=0)
 
     listen_parser = sub_parsers.add_parser(
         _LISTEN,
         help="listen mode - listen until KeyboardInterrupt",
         parents=[parent_parser, consumer_parser],
     )
-    listen_parser.add_argument(
-        "-p", "--partition", required=False, type=int, default=None
-    )
+    listen_parser.add_argument("-p", "--partition", required=False, type=int, default=None)
 
     if len(sys.argv) == 1:
         parser.print_help()

src/saluki/utils.py

@@ -1,14 +1,12 @@
 import datetime
 import logging
-from typing import Tuple, List
+from typing import List, Tuple
 
 from confluent_kafka import Message
-
 from streaming_data_types import DESERIALISERS
 from streaming_data_types.exceptions import ShortBufferException
 from streaming_data_types.utils import get_schema
 
-
 logger = logging.getLogger("saluki")
 
 
@@ -29,9 +27,7 @@ def __try_to_deserialise_message(payload: bytes) -> Tuple[str | None, str | None
         def fallback_deserialiser(payload: bytes) -> str:
             return payload.decode()
 
-        deserialiser = (
-            fallback_deserialiser  # Fall back to this if we need to so data isn't lost
-        )
+        deserialiser = fallback_deserialiser  # Fall back to this if we need to so data isn't lost
 
     logger.debug(f"Deserialiser: {deserialiser}")
 

tests/test_listen.py

@@ -1,10 +1,9 @@
-from unittest.mock import MagicMock
 
-from confluent_kafka import TopicPartition
 
-from saluki.listen import listen
 from unittest import mock
 
+from saluki.listen import listen
+
 
 def test_listen_with_partition_assigns_to_partition():
     expected_partition = 123

tests/test_utils.py

@@ -1,18 +1,18 @@
 from unittest.mock import Mock, patch
 
 import pytest
+from confluent_kafka import Message
 from streaming_data_types.forwarder_config_update_fc00 import (
     ConfigurationUpdate,
     StreamInfo,
 )
 
 from saluki.utils import (
-    parse_kafka_uri,
-    _parse_timestamp,
-    _deserialise_and_print_messages,
     __try_to_deserialise_message,
+    _deserialise_and_print_messages,
+    _parse_timestamp,
+    parse_kafka_uri,
 )
-from confluent_kafka import Message
 
 
 @pytest.fixture