chore: better error messages + bump version (#172)

Signed-off-by: Avik Basu <[email protected]>

pynumaflow/reducer/async_server.py

@@ -1,4 +1,5 @@
 import inspect
+from typing import Optional
 
 import aiorun
 import grpc
@@ -21,10 +22,12 @@
     Reducer,
 )
 
-from pynumaflow.shared.server import NumaflowServer, checkInstance, start_async_server
+from pynumaflow.shared.server import NumaflowServer, check_instance, start_async_server
 
 
-def get_handler(reducer_handler: ReduceCallable, init_args: tuple = (), init_kwargs: dict = None):
+def get_handler(
+    reducer_handler: ReduceCallable, init_args: tuple = (), init_kwargs: Optional[dict] = None
+):
     """
     Get the correct handler type based on the arguments passed
     """
@@ -35,13 +38,16 @@ def get_handler(reducer_handler: ReduceCallable, init_args: tuple = (), init_kwa
             raise TypeError("Cannot pass function handler with init args or kwargs")
         # return the function handler
         return reducer_handler
-    elif not checkInstance(reducer_handler, Reducer) and issubclass(reducer_handler, Reducer):
+    elif not check_instance(reducer_handler, Reducer) and issubclass(reducer_handler, Reducer):
         # if handler is type of Class Reducer, create a new instance of
         # a ReducerBuilderClass
         return _ReduceBuilderClass(reducer_handler, init_args, init_kwargs)
     else:
-        _LOGGER.error("Invalid Type: please provide the handler or the class name")
-        raise TypeError("Inavlid Type: please provide the handler or the class name")
+        _LOGGER.error(
+            _error_msg := f"Invalid Class Type {reducer_handler}: "
+            f"Please make sure the class type is passed, and it is a subclass of Reducer"
+        )
+        raise TypeError(_error_msg)
 
 
 class ReduceAsyncServer(NumaflowServer):

pynumaflow/reducestreamer/async_server.py

@@ -22,7 +22,7 @@
     ReduceStreamer,
 )
 
-from pynumaflow.shared.server import NumaflowServer, checkInstance, start_async_server
+from pynumaflow.shared.server import NumaflowServer, check_instance, start_async_server
 
 
 def get_handler(
@@ -38,15 +38,18 @@ def get_handler(
             raise TypeError("Cannot pass function handler with init args or kwargs")
         # return the function handler
         return reducer_handler
-    elif not checkInstance(reducer_handler, ReduceStreamer) and issubclass(
+    elif not check_instance(reducer_handler, ReduceStreamer) and issubclass(
         reducer_handler, ReduceStreamer
     ):
         # if handler is type of Class ReduceStreamer, create a new instance of
         # a ReducerBuilderClass
         return _ReduceStreamBuilderClass(reducer_handler, init_args, init_kwargs)
     else:
-        _LOGGER.error("Invalid Type: please provide the handler or the class name")
-        raise TypeError("Invalid Type: please provide the handler or the class name")
+        _LOGGER.error(
+            _error_msg := f"Invalid Class Type {reducer_handler}: "
+            f"Please make sure the class type is passed, and it is a subclass of ReduceStreamer"
+        )
+        raise TypeError(_error_msg)
 
 
 class ReduceStreamAsyncServer(NumaflowServer):

pynumaflow/shared/server.py

@@ -245,7 +245,7 @@ def _reserve_port(port_num: int) -> Iterator[int]:
         sock.close()
 
 
-def checkInstance(instance, callable_type) -> bool:
+def check_instance(instance, callable_type) -> bool:
     """
     Check if the given instance is of the given callable_type.
     """

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pynumaflow"
-version = "0.7.1"
+version = "0.7.2"
 description = "Provides the interfaces of writing Python User Defined Functions and Sinks for NumaFlow."
 authors = ["NumaFlow Developers"]
 readme = "README.md"

tests/reduce/test_async_reduce.py

@@ -8,8 +8,8 @@
 from google.protobuf import empty_pb2 as _empty_pb2
 from grpc.aio._server import Server
 
-from pynumaflow import setup_logging
 from pynumaflow._constants import WIN_START_TIME, WIN_END_TIME
+from pynumaflow.proto.reducer import reduce_pb2, reduce_pb2_grpc
 from pynumaflow.reducer import (
     Messages,
     Message,
@@ -18,15 +18,15 @@
     ReduceAsyncServer,
     Reducer,
 )
-from pynumaflow.proto.reducer import reduce_pb2, reduce_pb2_grpc
 from tests.testing_utils import (
     mock_message,
     mock_interval_window_start,
     mock_interval_window_end,
     get_time_args,
 )
 
-LOGGER = setup_logging(__name__)
+logging.basicConfig(level=logging.DEBUG)
+LOGGER = logging.getLogger(__name__)
 
 
 def request_generator(count, request, resetkey: bool = False):
@@ -243,6 +243,13 @@ def test_error_init(self):
         with self.assertRaises(TypeError):
             ReduceAsyncServer(ExampleClass(0))
 
+        # Check that an invalid class is passed
+        class ExampleBadClass:
+            pass
+
+        with self.assertRaises(TypeError):
+            ReduceAsyncServer(reducer_handler=ExampleBadClass)
+
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.DEBUG)

tests/reducestreamer/test_async_reduce.py

@@ -274,6 +274,13 @@ def test_error_init(self):
         with self.assertRaises(TypeError):
             ReduceStreamAsyncServer(ExampleClass(0))
 
+        # Check that an invalid class is passed
+        class ExampleBadClass:
+            pass
+
+        with self.assertRaises(TypeError):
+            ReduceStreamAsyncServer(reduce_stream_handler=ExampleBadClass)
+
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.DEBUG)