format all files

Signed-off-by: Sreekanth <prsreekanth920@gmail.com>

Author: BulkBeing

packages/pynumaflow-lite/manifests/accumulator/accumulator_stream_sorter.py

@@ -126,5 +126,6 @@ async def main():
         server.stop()
         return
 
+
 if __name__ == "__main__":
     asyncio.run(main())

packages/pynumaflow-lite/manifests/batchmap/batchmap_cat.py

@@ -7,9 +7,7 @@
 
 
 class SimpleBatchCat(batchmapper.BatchMapper):
-    async def handler(
-        self, batch: AsyncIterable[batchmapper.Datum]
-    ) -> batchmapper.BatchResponses:
+    async def handler(self, batch: AsyncIterable[batchmapper.Datum]) -> batchmapper.BatchResponses:
         responses = batchmapper.BatchResponses()
         async for d in batch:
             resp = batchmapper.BatchResponse(d.id)
@@ -21,10 +19,9 @@ async def handler(
             responses.append(resp)
         return responses
 
+
 async def start(
-    f: Callable[
-        [AsyncIterable[batchmapper.Datum]], Awaitable[batchmapper.BatchResponses]
-    ],
+    f: Callable[[AsyncIterable[batchmapper.Datum]], Awaitable[batchmapper.BatchResponses]],
 ):
     server = batchmapper.BatchMapAsyncServer()
 

packages/pynumaflow-lite/manifests/map/map_cat.py

@@ -17,6 +17,7 @@ async def handler(self, keys: list[str], payload: mapper.Datum) -> mapper.Messag
 
         return messages
 
+
 async def start(f: Callable[[list[str], mapper.Datum], Awaitable[mapper.Messages]]):
     server = mapper.MapAsyncServer()
 

packages/pynumaflow-lite/manifests/mapstream/mapstream_cat.py

@@ -7,16 +7,15 @@
 
 
 class SimpleStreamCat(mapstreamer.MapStreamer):
-    async def handler(
-        self, keys: list[str], datum: mapstreamer.Datum
-    ) -> AsyncIterator[Message]:
+    async def handler(self, keys: list[str], datum: mapstreamer.Datum) -> AsyncIterator[Message]:
         parts = datum.value.decode("utf-8").split(",")
         if not parts:
             yield Message.to_drop()
             return
         for s in parts:
             yield Message(s.encode(), keys)
 
+
 async def start(f: Callable[[list[str], mapstreamer.Datum], AsyncIterator[Message]]):
     # Use default socket/info file locations; no explicit sock file passed
     server = mapstreamer.MapStreamAsyncServer()

packages/pynumaflow-lite/manifests/reduce/reduce_counter_class.py

@@ -19,13 +19,12 @@ async def handler(
         self.counter = 0
         async for _ in datums:
             self.counter += 1
-        msg = (
-            f"counter:{self.counter} interval_window_start:{iw.start} interval_window_end:{iw.end}"
-        ).encode()
+        msg = (f"counter:{self.counter} interval_window_start:{iw.start} interval_window_end:{iw.end}").encode()
         out = reducer.Messages()
         out.append(reducer.Message(msg, keys))
         return out
 
+
 async def start(creator: type[reducer.Reducer], init_args: tuple):
     sock_file = "/var/run/numaflow/reduce.sock"
     server_info_file = "/var/run/numaflow/reducer-server-info"

packages/pynumaflow-lite/manifests/reducestream/reducestream_counter.py

@@ -53,23 +53,20 @@ async def handler(
             # Emit intermediate result every 10 items
             if self.counter % 10 == 0:
                 msg = (
-                    f"counter:{self.counter} "
-                    f"interval_window_start:{iw.start} "
-                    f"interval_window_end:{iw.end}"
+                    f"counter:{self.counter} interval_window_start:{iw.start} interval_window_end:{iw.end}"
                 ).encode()
                 print(f"Yielding intermediate result: counter={self.counter}")
                 # Early release of data - this is the key feature of reduce streaming!
                 yield reducestreamer.Message(msg, keys=keys)
 
         # Emit final result
         msg = (
-            f"counter:{self.counter} (FINAL) "
-            f"interval_window_start:{iw.start} "
-            f"interval_window_end:{iw.end}"
+            f"counter:{self.counter} (FINAL) interval_window_start:{iw.start} interval_window_end:{iw.end}"
         ).encode()
         print(f"Yielding final result: counter={self.counter}")
         yield reducestreamer.Message(msg, keys=keys)
 
+
 async def start(creator: type, init_args: tuple):
     """Start the reduce stream server."""
     sock_file = "/var/run/numaflow/reducestream.sock"

packages/pynumaflow-lite/manifests/session_reduce/session_reduce_counter_class.py

@@ -75,5 +75,6 @@ async def main():
         server.stop()
         return
 
+
 if __name__ == "__main__":
     asyncio.run(main())

packages/pynumaflow-lite/manifests/sideinput/sideinput_example.py

@@ -86,6 +86,7 @@ def update_data_from_file(self, path):
         except Exception as e:
             print(f"Error reading file: {e}")
 
+
 async def start_sideinput():
     """Start the SideInput retriever server."""
     server = sideinputer.SideInputAsyncServer()

packages/pynumaflow-lite/manifests/sink/sink_log.py

@@ -25,6 +25,7 @@ async def handler(self, datums: AsyncIterable[sinker.Datum]) -> sinker.Responses
             # we can use Response.as_fallback(msg.id) to write the message to fallback sink
         return responses
 
+
 async def start(
     f: Callable[[AsyncIterator[sinker.Datum]], Awaitable[sinker.Responses]],
 ):

packages/pynumaflow-lite/manifests/source/simple_source.py

@@ -22,15 +22,11 @@ def __init__(self):
         self.counter = 0
         self.partition_idx = 0
 
-    async def read_handler(
-        self, datum: sourcer.ReadRequest
-    ) -> AsyncIterator[sourcer.Message]:
+    async def read_handler(self, datum: sourcer.ReadRequest) -> AsyncIterator[sourcer.Message]:
         """
         The simple source generates messages with incrementing numbers.
         """
-        _LOGGER.info(
-            f"Read request: num_records={datum.num_records}, timeout_ms={datum.timeout_ms}"
-        )
+        _LOGGER.info(f"Read request: num_records={datum.num_records}, timeout_ms={datum.timeout_ms}")
 
         # Generate the requested number of messages
         for _ in range(datum.num_records):
@@ -66,19 +62,15 @@ async def ack_handler(self, request: sourcer.AckRequest) -> None:
         """
         _LOGGER.info(f"Acknowledging {len(request.offsets)} offsets")
         for offset in request.offsets:
-            _LOGGER.debug(
-                f"Acked offset: {offset.offset.decode('utf-8')}, partition: {offset.partition_id}"
-            )
+            _LOGGER.debug(f"Acked offset: {offset.offset.decode('utf-8')}, partition: {offset.partition_id}")
 
     async def nack_handler(self, request: sourcer.NackRequest) -> None:
         """
         The simple source negatively acknowledges the offsets.
         """
         _LOGGER.info(f"Negatively acknowledging {len(request.offsets)} offsets")
         for offset in request.offsets:
-            _LOGGER.warning(
-                f"Nacked offset: {offset.offset.decode('utf-8')}, partition: {offset.partition_id}"
-            )
+            _LOGGER.warning(f"Nacked offset: {offset.offset.decode('utf-8')}, partition: {offset.partition_id}")
 
     async def pending_handler(self) -> sourcer.PendingResponse:
         """
@@ -92,6 +84,7 @@ async def partitions_handler(self) -> sourcer.PartitionsResponse:
         """
         return sourcer.PartitionsResponse(partitions=[self.partition_idx])
 
+
 async def start():
     server = sourcer.SourceAsyncServer()