fix all ruff lints

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

Author: BulkBeing

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

@@ -7,8 +7,8 @@
 
 import asyncio
 import signal
+from collections.abc import AsyncIterator
 from datetime import datetime
-from typing import AsyncIterator
 
 from pynumaflow_lite.accumulator import (
     Accumulator,
@@ -107,7 +107,6 @@ async def main():
     """
     Start the accumulator server.
     """
-    import signal
 
     server = AccumulatorAsyncServer()
 
@@ -124,10 +123,7 @@ async def main():
         await server.start(StreamSorter)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 if __name__ == "__main__":

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

@@ -1,7 +1,6 @@
 import asyncio
 import signal
-from collections.abc import AsyncIterable
-from typing import Awaitable, Callable
+from collections.abc import AsyncIterable, Awaitable, Callable
 
 from pynumaflow_lite import batchmapper
 from pynumaflow_lite.batchmapper import Message
@@ -41,10 +40,7 @@ async def start(
         await server.start(f)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 

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

@@ -1,6 +1,6 @@
 import asyncio
 import signal
-from typing import Awaitable, Callable
+from collections.abc import Awaitable, Callable
 
 from pynumaflow_lite import mapper
 
@@ -38,10 +38,7 @@ async def start(f: Callable[[list[str], mapper.Datum], Awaitable[mapper.Messages
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
         # Fallback in case the task was cancelled by the runner
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 

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

@@ -1,7 +1,6 @@
 import asyncio
 import signal
-from collections.abc import AsyncIterator
-from typing import Callable
+from collections.abc import AsyncIterator, Callable
 
 from pynumaflow_lite import mapstreamer
 from pynumaflow_lite.mapstreamer import Message
@@ -34,10 +33,7 @@ async def start(f: Callable[[list[str], mapstreamer.Datum], AsyncIterator[Messag
         await server.start(f)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 

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

@@ -42,10 +42,7 @@ async def start(creator: type[reducer.Reducer], init_args: tuple):
         await server.start(creator, init_args)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 

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

@@ -88,10 +88,7 @@ async def start(creator: type, init_args: tuple):
         await server.start(creator, init_args)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 

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

@@ -72,10 +72,7 @@ async def main():
         await server.start(SessionReduceCounter)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 if __name__ == "__main__":

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

@@ -14,9 +14,10 @@
 import threading
 from threading import Thread
 
-from pynumaflow_lite import mapper, sideinputer
 from watchfiles import watch
 
+from pynumaflow_lite import mapper, sideinputer
+
 
 class ExampleSideInput(sideinputer.SideInput):
     """
@@ -32,7 +33,7 @@ async def retrieve_handler(self) -> sideinputer.Response:
         """
         time_now = datetime.datetime.now()
         # val is the value to be broadcasted
-        val = f"an example: {str(time_now)}"
+        val = f"an example: {time_now!s}"
         self.counter += 1
         # broadcast_message() is used to indicate that there is a broadcast
         return sideinputer.Response.broadcast_message(val.encode("utf-8"))
@@ -65,7 +66,7 @@ def file_watcher(self):
         path = sideinputer.DIR_PATH
         for changes in watch(path):
             for change in changes:
-                change_type, file_path = change
+                _change_type, file_path = change
                 if file_path.endswith(self.watched_file):
                     with self.data_value_lock:
                         self.update_data_from_file(file_path)

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

@@ -1,8 +1,7 @@
 import asyncio
 import logging
 import signal
-from collections.abc import AsyncIterable, AsyncIterator
-from typing import Awaitable, Callable
+from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable
 
 from pynumaflow_lite import sinker
 from pynumaflow_lite._sink_dtypes import Sinker
@@ -43,10 +42,7 @@ async def start(
         await server.start(f)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return
 
 

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

@@ -33,7 +33,7 @@ async def read_handler(
         )
 
         # Generate the requested number of messages
-        for i in range(datum.num_records):
+        for _ in range(datum.num_records):
             # Create message payload
             payload = f"message-{self.counter}".encode()
 
@@ -110,10 +110,7 @@ async def start():
         await server.start(handler)
         print("Shutting down gracefully...")
     except asyncio.CancelledError:
-        try:
-            server.stop()
-        except Exception:
-            pass
+        server.stop()
         return