Remove the need for a mapper/sinker superclass. Just use callables

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

Author: BulkBeing

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

@@ -1,10 +1,10 @@
 from collections.abc import AsyncIterator
 
 from pynumaflow_lite import batchmapper
-from pynumaflow_lite.batchmapper import BatchMapper, BatchResponse, Datum, Message
+from pynumaflow_lite.batchmapper import BatchResponse, Datum, Message
 
 
-class SimpleBatchCat(BatchMapper):
+class SimpleBatchCat:
     async def handler(self, batch: AsyncIterator[Datum]) -> list[BatchResponse]:
         return [
             BatchResponse(
@@ -16,4 +16,5 @@ async def handler(self, batch: AsyncIterator[Datum]) -> list[BatchResponse]:
 
 
 if __name__ == "__main__":
-    batchmapper.BatchMapAsyncServer(SimpleBatchCat()).run()
+    batch_mapper_obj = SimpleBatchCat()
+    batchmapper.BatchMapAsyncServer(batch_mapper_obj.handler).run()

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

@@ -1,12 +1,13 @@
 from pynumaflow_lite import mapper
 
 
-class SimpleCat(mapper.Mapper):
+class SimpleCat:
     async def handler(self, datum: mapper.Datum) -> list[mapper.Message]:
         if datum.value == b"bad world":
             return [mapper.Message.to_drop()]
         return [mapper.Message(datum.value, keys=datum.keys)]
 
 
 if __name__ == "__main__":
-    mapper.MapAsyncServer(SimpleCat()).run()
+    mapper_obj = SimpleCat()
+    mapper.MapAsyncServer(mapper_obj.handler).run()

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

@@ -4,7 +4,7 @@
 from pynumaflow_lite.mapstreamer import Message
 
 
-class SimpleStreamCat(mapstreamer.MapStreamer):
+class SimpleStreamCat:
     async def handler(self, datum: mapstreamer.Datum) -> AsyncIterable[Message]:
         parts = datum.value.decode("utf-8").split(",")
         if not parts:
@@ -15,4 +15,5 @@ async def handler(self, datum: mapstreamer.Datum) -> AsyncIterable[Message]:
 
 
 if __name__ == "__main__":
-    mapstreamer.MapStreamAsyncServer(SimpleStreamCat()).run()
+    map_streamer_obj = SimpleStreamCat()
+    mapstreamer.MapStreamAsyncServer(map_streamer_obj.handler).run()

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

@@ -39,7 +39,7 @@ async def retrieve_handler(self) -> sideinputer.Response:
         return sideinputer.Response.broadcast_message(val.encode("utf-8"))
 
 
-class SideInputHandler(mapper.Mapper):
+class SideInputHandler:
     """
     A Mapper that reads from side input files and includes the value in its output.
     """
@@ -103,16 +103,16 @@ async def start_sideinput():
 
 def start_mapper():
     """Start the Mapper server that reads from side inputs."""
-    handler = SideInputHandler()
+    mapper_obj = SideInputHandler()
 
     # Initialize the data value from the side input file
-    handler.init_data_value()
+    mapper_obj.init_data_value()
 
     # Start the file watcher in a background thread
-    watcher_thread = Thread(target=handler.file_watcher, daemon=True)
+    watcher_thread = Thread(target=mapper_obj.file_watcher, daemon=True)
     watcher_thread.start()
 
-    mapper.MapAsyncServer(handler).run()
+    mapper.MapAsyncServer(mapper_obj.handler).run()
 
 
 if __name__ == "__main__":

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

@@ -1,20 +1,19 @@
 import logging
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterator
 
 from pynumaflow_lite import sinker
-from pynumaflow_lite.sinker import Sinker
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 _LOGGER = logging.getLogger(__name__)
 
 
-class SimpleLogSink(Sinker):
+class SimpleLogSink:
     """
     Simple log sink that logs each message and returns success responses.
     """
 
-    async def handler(self, datums: AsyncIterable[sinker.Datum]) -> list[sinker.Response]:
+    async def handler(self, datums: AsyncIterator[sinker.Datum]) -> list[sinker.Response]:
         responses = []
         async for msg in datums:
             _LOGGER.info("User Defined Sink: %s", msg.value.decode("utf-8"))
@@ -25,4 +24,5 @@ async def handler(self, datums: AsyncIterable[sinker.Datum]) -> list[sinker.Resp
 
 
 if __name__ == "__main__":
-    sinker.SinkAsyncServer(SimpleLogSink()).run()
+    sinker_obj = SimpleLogSink()
+    sinker.SinkAsyncServer(sinker_obj.handler).run()

packages/pynumaflow-lite/pynumaflow_lite/__init__.py

@@ -61,34 +61,26 @@
 except Exception:  # pragma: no cover
     sideinputer = None
 
-# Surface the Python Mapper, BatchMapper, MapStreamer, Reducer, SessionReducer, ReduceStreamer, Accumulator, Sinker,
-# Sourcer, SourceTransformer, and SideInput classes under the extension submodules for convenient access
+# Surface the Python async servers and data-type classes under the extension submodules for convenient access
 from ._accumulator_dtypes import Accumulator
 from ._batchmap_server import BatchMapAsyncServer
-from ._batchmapper_dtypes import BatchMapper
-from ._map_dtypes import Mapper
 from ._map_server import MapAsyncServer
-from ._mapstream_dtypes import MapStreamer
 from ._mapstream_server import MapStreamAsyncServer
 from ._reduce_dtypes import Reducer
 from ._reducestreamer_dtypes import ReduceStreamer
 from ._session_reduce_dtypes import SessionReducer
 from ._sideinput_dtypes import SideInput
-from ._sink_dtypes import Sinker
 from ._sink_server import SinkAsyncServer
 from ._source_dtypes import Sourcer
 from ._sourcetransformer_dtypes import SourceTransformer
 
 if mapper is not None:
-    mapper.Mapper = Mapper
     mapper.MapAsyncServer = MapAsyncServer
 
 if batchmapper is not None:
-    batchmapper.BatchMapper = BatchMapper
     batchmapper.BatchMapAsyncServer = BatchMapAsyncServer
 
 if mapstreamer is not None:
-    mapstreamer.MapStreamer = MapStreamer
     mapstreamer.MapStreamAsyncServer = MapStreamAsyncServer
 
 if reducer is not None:
@@ -104,7 +96,6 @@
     accumulator.Accumulator = Accumulator
 
 if sinker is not None:
-    sinker.Sinker = Sinker
     sinker.SinkAsyncServer = SinkAsyncServer
 
 if sourcer is not None:

packages/pynumaflow-lite/pynumaflow_lite/_batchmap_server.py

@@ -2,16 +2,22 @@
 
 import asyncio
 import signal
+from collections.abc import AsyncIterator, Awaitable, Callable
 from types import TracebackType
-from typing import Any
 
 from .pynumaflow_lite import batchmapper as _batchmapper
 
+BatchResponse = _batchmapper.BatchResponse
+Datum = _batchmapper.Datum
+
 
 class BatchMapAsyncServer:
     def __init__(
         self,
-        handler: Any,
+        handler: Callable[
+            [AsyncIterator[Datum]],
+            Awaitable[list[BatchResponse]],
+        ],
         *,
         sock_file: str | None = None,
         server_info_file: str | None = None,

packages/pynumaflow-lite/pynumaflow_lite/_batchmapper_dtypes.py

@@ -2,16 +2,19 @@
 
 import asyncio
 import signal
+from collections.abc import Awaitable, Callable
 from types import TracebackType
-from typing import Any
 
 from .pynumaflow_lite import mapper as _mapper
 
+Datum = _mapper.Datum
+Message = _mapper.Message
+
 
 class MapAsyncServer:
     def __init__(
         self,
-        handler: Any,
+        handler: Callable[[Datum], Awaitable[list[Message]]],
         *,
         sock_file: str | None = None,
         server_info_file: str | None = None,