chore: linting and static checks for pynumaflow-lite (#346)

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

Author: vaibhavtiwari33

packages/pynumaflow-lite/Makefile

@@ -43,11 +43,17 @@ test-rust:
 clean:
 	cargo clean
 
+py-fmt:
+	uv run black pynumaflow_lite/ tests/ manifests/
+
+py-lint: py-fmt
+	uv run ruff check --fix .
+
 fmt:
 	cargo fmt --all
 
 .PHONY: lint
-lint: test-fmt clippy
+lint: test-fmt clippy py-lint
 
 .PHONY: test-fmt
 test-fmt:

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

@@ -4,11 +4,18 @@
 This accumulator buffers incoming data and sorts it by event time,
 flushing sorted data when the watermark advances.
 """
+
+import signal
 import asyncio
 from datetime import datetime
 from typing import AsyncIterator
 
-from pynumaflow_lite.accumulator import Datum, Message, AccumulatorAsyncServer, Accumulator
+from pynumaflow_lite.accumulator import (
+    Datum,
+    Message,
+    AccumulatorAsyncServer,
+    Accumulator,
+)
 
 
 class StreamSorter(Accumulator):
@@ -19,6 +26,7 @@ class StreamSorter(Accumulator):
 
     def __init__(self):
         from datetime import timezone
+
         # Initialize with a very old timestamp (timezone-aware)
         self.latest_wm = datetime.fromtimestamp(-1, tz=timezone.utc)
         self.sorted_buffer: list[Datum] = []
@@ -33,8 +41,10 @@ async def handler(self, datums: AsyncIterator[Datum]) -> AsyncIterator[Message]:
 
         async for datum in datums:
             datum_count += 1
-            print(f"Received datum #{datum_count}: event_time={datum.event_time}, "
-                  f"watermark={datum.watermark}, value={datum.value}")
+            print(
+                f"Received datum #{datum_count}: event_time={datum.event_time}, "
+                f"watermark={datum.watermark}, value={datum.value}"
+            )
 
             # If watermark has moved forward
             if datum.watermark and datum.watermark > self.latest_wm:
@@ -122,7 +132,7 @@ async def main():
 
 
 # Optional: ensure default signal handlers are in place so asyncio.run can handle them cleanly.
-import signal
+
 signal.signal(signal.SIGINT, signal.default_int_handler)
 try:
     signal.signal(signal.SIGTERM, signal.SIG_DFL)

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

@@ -8,7 +8,9 @@
 
 
 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)
@@ -29,7 +31,11 @@ async def handler(self, batch: AsyncIterable[batchmapper.Datum]) -> batchmapper.
     pass
 
 
-async def start(f: Callable[[AsyncIterable[batchmapper.Datum]], Awaitable[batchmapper.BatchResponses]]):
+async def start(
+    f: Callable[
+        [AsyncIterable[batchmapper.Datum]], Awaitable[batchmapper.BatchResponses]
+    ],
+):
     server = batchmapper.BatchMapAsyncServer()
 
     # Register loop-level signal handlers so we control shutdown and avoid asyncio.run

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

@@ -6,9 +6,7 @@
 
 
 class SimpleCat(mapper.Mapper):
-    async def handler(
-            self, keys: list[str], payload: mapper.Datum
-    ) -> mapper.Messages:
+    async def handler(self, keys: list[str], payload: mapper.Datum) -> mapper.Messages:
 
         messages = mapper.Messages()
 

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

@@ -8,7 +8,9 @@
 
 
 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()
@@ -51,4 +53,3 @@ async def start(f: Callable[[list[str], mapstreamer.Datum], AsyncIterator[Messag
 if __name__ == "__main__":
     async_handler = SimpleStreamCat()
     asyncio.run(start(async_handler))
-

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

@@ -10,7 +10,10 @@ def __init__(self, initial: int = 0) -> None:
         self.counter = initial
 
     async def handler(
-        self, keys: list[str], datums: AsyncIterable[reducer.Datum], md: reducer.Metadata
+        self,
+        keys: list[str],
+        datums: AsyncIterable[reducer.Datum],
+        md: reducer.Metadata,
     ) -> reducer.Messages:
         iw = md.interval_window
         self.counter = 0
@@ -57,4 +60,3 @@ async def start(creator: type[reducer.Reducer], init_args: tuple):
 
 if __name__ == "__main__":
     asyncio.run(start(ReduceCounter, (0,)))
-

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

@@ -7,6 +7,7 @@
 The counter increments for each datum and emits a message every 10 items,
 plus a final message at the end.
 """
+
 import asyncio
 import signal
 from collections.abc import AsyncIterable, AsyncIterator
@@ -17,12 +18,12 @@
 class ReduceCounter(reducestreamer.ReduceStreamer):
     """
     A reduce streaming counter that emits intermediate results.
-    
+
     This demonstrates the key difference from regular Reducer:
     - Regular Reducer: waits for all data, then returns Messages
     - ReduceStreamer: yields Message objects incrementally as an async iterator
     """
-    
+
     def __init__(self, initial: int = 0) -> None:
         self.counter = initial
 
@@ -34,21 +35,21 @@ async def handler(
     ) -> AsyncIterator[reducestreamer.Message]:
         """
         Process datums and yield messages incrementally.
-        
+
         Args:
             keys: List of keys for this window
             datums: Async iterable of incoming data
             md: Metadata containing window information
-            
+
         Yields:
             Message objects to send to the next vertex
         """
         iw = md.interval_window
         print(f"Handler started for keys={keys}, window=[{iw.start}, {iw.end}]")
-        
+
         async for _ in datums:
             self.counter += 1
-            
+
             # Emit intermediate result every 10 items
             if self.counter % 10 == 0:
                 msg = (
@@ -59,7 +60,7 @@ async def handler(
                 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) "
@@ -105,4 +106,3 @@ async def start(creator: type, init_args: tuple):
 
 if __name__ == "__main__":
     asyncio.run(start(ReduceCounter, (0,)))
-

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

@@ -25,7 +25,7 @@ def __init__(self, initial: int = 0) -> None:
         self.counter = initial
 
     async def session_reduce(
-            self, keys: list[str], datums: AsyncIterable[session_reducer.Datum]
+        self, keys: list[str], datums: AsyncIterable[session_reducer.Datum]
     ) -> AsyncIterator[session_reducer.Message]:
         """
         Count all incoming messages in this session and yield the count.

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

@@ -6,6 +6,7 @@
 - If MAPPER is set to "true", runs as a Mapper that reads side input files
 - Otherwise, runs as a SideInput retriever that broadcasts values
 """
+
 import asyncio
 import os
 import signal
@@ -142,4 +143,3 @@ async def start_mapper():
     else:
         print("Starting as SideInput retriever...")
         asyncio.run(start_sideinput())
-

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

@@ -1,5 +1,4 @@
 import asyncio
-import collections
 import logging
 import signal
 from collections.abc import AsyncIterable, AsyncIterator
@@ -36,7 +35,9 @@ async def handler(self, datums: AsyncIterable[sinker.Datum]) -> sinker.Responses
     pass
 
 
-async def start(f: Callable[[AsyncIterator[sinker.Datum]], Awaitable[sinker.Responses]]):
+async def start(
+    f: Callable[[AsyncIterator[sinker.Datum]], Awaitable[sinker.Responses]],
+):
     server = sinker.SinkAsyncServer()
 
     # Register loop-level signal handlers so we control shutdown and avoid asyncio.run
@@ -61,4 +62,3 @@ async def start(f: Callable[[AsyncIterator[sinker.Datum]], Awaitable[sinker.Resp
 if __name__ == "__main__":
     async_handler = SimpleLogSink()
     asyncio.run(start(async_handler))
-