ruff

Author: rerpha

src/saluki/consume.py

@@ -16,7 +16,7 @@ def consume(
     offset: int | None = None,
     go_forwards: bool = False,
     schemas_to_filter_to: list[str] | None = None,
-    timestamp: int|None = None,
+    timestamp: int | None = None,
 ) -> None:
     """
     consume from a topic and deserialise each message
@@ -44,9 +44,7 @@ def consume(
     )
 
     if timestamp is not None:
-        offset = c.offsets_for_times(
-            [TopicPartition(topic, partition, timestamp)]
-        )[0].offset
+        offset = c.offsets_for_times([TopicPartition(topic, partition, timestamp)])[0].offset
         logger.debug(f"offset for timestamp {timestamp} is {offset}")
 
     if go_forwards:

src/saluki/main.py

@@ -6,7 +6,7 @@
 from saluki.listen import listen
 from saluki.play import play
 from saluki.sniff import sniff
-from saluki.utils import parse_kafka_uri, dateutil_parsable_or_unix_timestamp
+from saluki.utils import dateutil_parsable_or_unix_timestamp, parse_kafka_uri
 
 logger = logging.getLogger("saluki")
 logging.basicConfig(level=logging.INFO)
@@ -23,7 +23,7 @@ def main() -> None:
         description="serialise/de-serialise flatbuffers and consume/produce from/to kafka",
     )
     common_options = argparse.ArgumentParser(add_help=False)
-    common_options.add_argument("-v", "--verbose", help="show DEBUG logs", action='store_true')
+    common_options.add_argument("-v", "--verbose", help="show DEBUG logs", action="store_true")
     common_options.add_argument(
         "-l",
         "--log-file",
@@ -48,8 +48,12 @@ def main() -> None:
 
     sub_parsers = parser.add_subparsers(help="sub-command help", required=True, dest="command")
 
-    sniff_parser = sub_parsers.add_parser(_SNIFF, help="sniff - broker metadata", parents=[common_options])
-    sniff_parser.add_argument("broker", type=str, help="broker, optionally suffixed with a topic name to filter to")
+    sniff_parser = sub_parsers.add_parser(
+        _SNIFF, help="sniff - broker metadata", parents=[common_options]
+    )
+    sniff_parser.add_argument(
+        "broker", type=str, help="broker, optionally suffixed with a topic name to filter to"
+    )
 
     consumer_parser = argparse.ArgumentParser(add_help=False)
     consumer_parser.add_argument(
@@ -75,9 +79,17 @@ def main() -> None:
     consumer_mode_parser.add_argument("-g", "--go-forwards", required=False, action="store_true")
     cg = consumer_mode_parser.add_mutually_exclusive_group(required=False)
     cg.add_argument(
-        "-o", "--offset", help="offset to consume from", type=int,
+        "-o",
+        "--offset",
+        help="offset to consume from",
+        type=int,
+    )
+    cg.add_argument(
+        "-t",
+        "--timestamp",
+        help="timestamp to consume from",
+        type=dateutil_parsable_or_unix_timestamp,
     )
-    cg.add_argument("-t", "--timestamp", help="timestamp to consume from", type=dateutil_parsable_or_unix_timestamp)
 
     listen_parser = sub_parsers.add_parser(  # noqa: F841
         _LISTEN,
@@ -99,7 +111,14 @@ def main() -> None:
         type=int,
         nargs=2,
     )
-    g.add_argument("-t", "--timestamps", help='timestamps to replay between in ISO8601 or RFC3339 format ie. "2025-11-17 07:00:00 or as a unix timestamp"  ', type=dateutil_parsable_or_unix_timestamp, nargs=2)
+    g.add_argument(
+        "-t",
+        "--timestamps",
+        help="timestamps to replay between in ISO8601 or RFC3339 format ie."
+        ' "2025-11-17 07:00:00 or as a unix timestamp"  ',
+        type=dateutil_parsable_or_unix_timestamp,
+        nargs=2,
+    )
 
     if len(sys.argv) == 1:
         parser.print_help()
@@ -121,7 +140,14 @@ def main() -> None:
     elif args.command == _CONSUME:
         broker, topic = parse_kafka_uri(args.topic)
         consume(
-            broker, topic, args.partition, args.messages, args.offset, args.go_forwards, args.filter, args.timestamp
+            broker,
+            topic,
+            args.partition,
+            args.messages,
+            args.offset,
+            args.go_forwards,
+            args.filter,
+            args.timestamp,
         )
     elif args.command == _PLAY:
         src_broker, src_topic = parse_kafka_uri(args.topics[0])

src/saluki/play.py

@@ -1,5 +1,6 @@
 import logging
 import uuid
+
 from confluent_kafka import Consumer, Producer, TopicPartition
 
 logger = logging.getLogger("saluki")
@@ -44,11 +45,13 @@ def play(
         start_offset = consumer.offsets_for_times(
             [
                 TopicPartition(src_topic, src_partition, timestamps[0]),
-
             ]
         )[0]
-        # See https://github.com/confluentinc/confluent-kafka-python/issues/1178 as to why offsets_for_times is called twice.
-        stop_offset = consumer.offsets_for_times([TopicPartition(src_topic, src_partition, timestamps[1])])[0]
+        # See https://github.com/confluentinc/confluent-kafka-python/issues/1178
+        # as to why offsets_for_times is called twice.
+        stop_offset = consumer.offsets_for_times(
+            [TopicPartition(src_topic, src_partition, timestamps[1])]
+        )[0]
     elif offsets is not None:
         start_offset = TopicPartition(src_topic, src_partition, offsets[0])
         stop_offset = TopicPartition(src_topic, src_partition, offsets[1])
@@ -66,7 +69,9 @@ def play(
         msgs = consumer.consume(num_messages)
         logger.debug(f"finished consuming {num_messages} messages")
         consumer.close()
-        producer.produce_batch(dest_topic, [{'key': message.key(), 'value': message.value()} for message in msgs])
+        producer.produce_batch(
+            dest_topic, [{"key": message.key(), "value": message.value()} for message in msgs]
+        )
         logger.debug(f"flushing producer. len(p): {len(producer)}")
         producer.flush(timeout=10)
 

src/saluki/utils.py

@@ -5,7 +5,7 @@
 from zoneinfo import ZoneInfo
 
 from confluent_kafka import Message
-from dateutil.parser import parse, ParserError
+from dateutil.parser import ParserError, parse
 from streaming_data_types import DESERIALISERS
 from streaming_data_types.exceptions import ShortBufferException
 from streaming_data_types.utils import get_schema
@@ -102,7 +102,11 @@ def dateutil_parsable_or_unix_timestamp(inp: str) -> int:
         try:
             return int(round(parse(inp).timestamp() * 1000))
         except (ParserError, OverflowError):
-            logger.debug(f"Failed to parse {inp} as a dateutil parsable. Falling back to unix timestamp")
+            logger.debug(
+                f"Failed to parse {inp} as a dateutil parsable. Falling back to unix timestamp"
+            )
             return int(inp)
     except ValueError:
-        raise ArgumentTypeError(f"timestamp {inp} is not parsable by dateutil.parse() and is not a unix timestamp")
+        raise ArgumentTypeError(
+            f"timestamp {inp} is not parsable by dateutil.parse() and is not a unix timestamp"
+        )

tests/test_sniff.py

@@ -1,10 +1,11 @@
-import pytest
 from unittest.mock import patch
 
+import pytest
 from confluent_kafka.admin import BrokerMetadata, ClusterMetadata, TopicMetadata
 
 from saluki.sniff import sniff
 
+
 @pytest.fixture()
 def fake_cluster_md():
     """
@@ -27,6 +28,7 @@ def fake_cluster_md():
     fake_cluster_md.topics = {"topic1": topic1, "topic2": topic2}
     return fake_cluster_md
 
+
 def test_sniff_with_two_partitions_in_a_topic(fake_cluster_md):
     with (
         patch("saluki.sniff.AdminClient") as a,
@@ -50,13 +52,13 @@ def test_sniff_with_two_partitions_in_a_topic(fake_cluster_md):
         topic2_call2 = logger.info.call_args_list[8]
         assert "1 - low:1, high:2, num_messages:1" in topic2_call2.args[0]
 
+
 def test_sniff_with_single_topic(fake_cluster_md):
     with (
         patch("saluki.sniff.AdminClient") as a,
         patch("saluki.sniff.Consumer") as c,
         patch("saluki.sniff.logger") as logger,
     ):
-
         a().list_topics.return_value = fake_cluster_md
         c().get_watermark_offsets.return_value = 1, 2
         sniff("mybroker:9093", "topic1")