ruff

Author: rerpha

src/saluki/consume.py

@@ -47,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/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

@@ -23,9 +23,7 @@ def _try_to_deserialise_message(payload: bytes) -> Tuple[str | None, str | None]
     def fallback_deserialiser(payload: bytes) -> str:
         return payload.decode()
 
-    deserialiser = DESERIALISERS.get(
-        schema if schema is not None else "", fallback_deserialiser
-    )
+    deserialiser = DESERIALISERS.get(schema if schema is not None else "", fallback_deserialiser)
     logger.debug(f"Deserialiser: {deserialiser}")
 
     ret = deserialiser(payload)

tests/test_utils.py

@@ -8,9 +8,9 @@
 )
 
 from saluki.utils import (
+    _parse_timestamp,
     _try_to_deserialise_message,
     deserialise_and_print_messages,
-    _parse_timestamp,
     parse_kafka_uri,
 )