Improve type handling on adding values to timeseries builder as well (#109)

Timeseries builder fixes:
- Add automatic type conversion, now supporting any number type (numpy.int64 for example) and not just built-in python types
- Null values are not throwing an exception anymore. They are just ignored

Author: harisbotic

src/PythonClient/src/quixstreams/builders/timeseriesdatabuilder.py

@@ -1,4 +1,5 @@
 import ctypes
+import numbers
 from typing import Union, Dict
 
 from ..helpers.nativedecorator import nativedecorator
@@ -27,26 +28,25 @@ def __init__(self, net_pointer: ctypes.c_void_p):
     def __enter__(self):
         self._entered = True
 
-    def add_value(self, parameter_id: str, value: Union[str, float, int, bytes, bytearray]) -> 'TimeseriesDataBuilder':
+    def add_value(self, parameter_id: str,  value: Union[numbers.Number, str, bytearray, bytes]) -> 'TimeseriesDataBuilder':
         """
         Adds new parameter value at the time the builder is created for.
 
         Args:
             parameter_id: The id of the parameter to set the value for.
-            value: The value of type string, float, int, bytes, or bytearray.
+            value: The value to add. Can be a number, string, bytearray, or bytes.
 
         Returns:
             The builder.
         """
 
-        val_type = type(value)
-        if val_type is int:
+        if value is None:
+            return self
+
+        if issubclass(type(value), numbers.Number):
             value = float(value)
-            val_type = float
-        elif val_type is bytearray:
-            value = bytes(value)
-            val_type = bytes
 
+        val_type = type(value)
         if val_type is float:
             new = tsdbi(self._interop.AddValue(parameter_id, value))
             if new != self._interop:
@@ -57,7 +57,7 @@ def add_value(self, parameter_id: str, value: Union[str, float, int, bytes, byte
             if new != self._interop:
                 self._interop.dispose_ptr__()
                 self._interop = new
-        elif val_type is bytes:
+        elif val_type is bytes or val_type is bytearray:
             arr_ptr = Array.WriteBytes(value)
             new = tsdbi(self._interop.AddValue3(parameter_id, arr_ptr))
             if new != self._interop:

src/PythonClient/tests/quixstreams/integrationtests/test_integration.py

@@ -3,6 +3,7 @@
 import unittest
 import threading
 import pandas as pd
+import numpy as np
 from src.quixstreams import Logging, LogLevel, AutoOffsetReset
 
 from testcontainers.core.container import DockerContainer
@@ -772,7 +773,7 @@ def on_stream_closed(stream: qx.StreamConsumer, end_type: qx.StreamEndType):
         print(f"---- Write second stream {output_stream.stream_id} ----")
 
         # Assert
-        self.waitforresult(event)
+        self.waitforresult(event, 30)
 
         self.assertEqual(last_stream_read.stream_id, output_stream.stream_id)
 
@@ -908,6 +909,23 @@ def test_get_or_create_stream_with_prev_stream_without_callback(self):
 # endregion
 
 # region timeseries data integration tests
+    def test_timeseries_builder_works_with_any_number_type_and_none(self):
+        # Arrange
+        print("Starting Integration test {}".format(sys._getframe().f_code.co_name))
+        topic_name = sys._getframe().f_code.co_name  # current method name
+        stream = qx.KafkaStreamingClient(TestIntegration.broker_list, None).get_topic_producer(topic_name).create_stream()
+
+        # Act
+        stream.timeseries.buffer \
+            .add_timestamp_nanoseconds(1) \
+            .add_value("npy_float64", np.float64(42.0)) \
+            .add_value("npy_int64", np.int64(42)) \
+            .add_value("native_int", int(42)) \
+            .add_value("native_float", float(42)) \
+            .add_value("none", None)
+
+        # Assert that no exception got raised
+
     def test_parameters_write_binary_read_binary_is_of_bytes(self):
         # Arrange
         print("Starting Integration test {}".format(sys._getframe().f_code.co_name))

src/PythonClient/tests/quixstreams/unittests/models/test_timeseriesdatatimestamp.py

@@ -8,56 +8,68 @@ class TimeseriesDataTimestampTests(unittest.TestCase):
 
     def test_add_double(self):
         # Arrange
-        pd = TimeseriesData()
+        td = TimeseriesData()
 
         # Act
-        pd.add_timestamp_nanoseconds(100) \
+        td.add_timestamp_nanoseconds(100) \
             .add_value("double", 1.232)
 
         # Assert
-        self.assertEqual(1.232, pd.timestamps[0].parameters["double"].numeric_value)
+        self.assertEqual(1.232, td.timestamps[0].parameters["double"].numeric_value)
 
     def test_add_string(self):
         # Arrange
-        pd = TimeseriesData()
+        td = TimeseriesData()
 
         # Act
-        pd.add_timestamp_nanoseconds(100) \
+        td.add_timestamp_nanoseconds(100) \
             .add_value("str", "value")
 
         # Assert
-        asdf = pd.timestamps[0]
+        asdf = td.timestamps[0]
         param = asdf.parameters
-        self.assertEqual(pd.timestamps[0].parameters["str"].string_value, "value")
+        self.assertEqual(td.timestamps[0].parameters["str"].string_value, "value")
 
     def test_add_bytes(self):
         # Arrange
-        pd = TimeseriesData()
+        td = TimeseriesData()
         expected = bytes("some bytes", "utf-8")
 
         # Act
-        pd.add_timestamp_nanoseconds(100) \
+        td.add_timestamp_nanoseconds(100) \
             .add_value("bytes", expected)
 
         # Assert
-        self.assertEqual(expected, pd.timestamps[0].parameters["bytes"].binary_value)
+        self.assertEqual(expected, td.timestamps[0].parameters["bytes"].binary_value)
+
+    def test_add_bytearray(self):
+        # Arrange
+        td = TimeseriesData()
+        expected = bytearray("some bytes", "utf-8")
+
+        # Act
+        td.add_timestamp_nanoseconds(100) \
+            .add_value("bytearray", expected)
+
+        # Assert
+        self.assertEqual(expected, td.timestamps[0].parameters["bytearray"].binary_value)
 
     def test_add_tag(self):
         # Arrange
-        pd = TimeseriesData()
-        pd.add_timestamp_nanoseconds(100)
-        pdts = pd.timestamps[0]
-        pdts.add_tag("a", "b")
+        td = TimeseriesData()
+        td.add_timestamp_nanoseconds(100)
+        tdts = td.timestamps[0]
+        tdts.add_tag("a", "b")
 
         # Act
-        pd.add_timestamp_nanoseconds(200)
-        pdts2 = pd.timestamps[1]
-        pdts.add_tag("b", "c")
-        pdts2.add_tags(pdts.tags)
+        td.add_timestamp_nanoseconds(200)
+        tdts2 = td.timestamps[1]
+        tdts.add_tag("b", "c")
+        tdts2.add_tags(tdts.tags)
 
         # Assert
-        self.assertEqual(pdts2.tags["a"], "b")
-        self.assertEqual(pdts2.tags["b"], "c")
+        self.assertEqual(tdts2.tags["a"], "b")
+        self.assertEqual(tdts2.tags["b"], "c")
 
     def test_add_tags(self):
         # Act