ruff

Author: rerpha

src/saluki/consume.py

@@ -47,7 +47,9 @@ 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,7 +21,9 @@ 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",
@@ -39,7 +41,9 @@ 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(
@@ -64,16 +68,24 @@ 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,7 +23,9 @@ 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)