chore: (WIP - 2 failing test) use kwargs in batching mode.

Author: karel-rehor

influxdb_client_3/__init__.py

@@ -377,8 +377,6 @@ def write(self, record=None, database=None, **kwargs):
         :type database: str
         :param kwargs: Additional arguments to pass to the write API.
         """
-        print("DEBUG InfluxDBClient3.write")
-        print(f"DEBUG kwargs {kwargs} ")
         if database is None:
             database = self._database
 

influxdb_client_3/write_client/client/write_api.py

@@ -164,10 +164,11 @@ def add_default_tag(self, key, value) -> None:
 
 
 class _BatchItemKey(object):
-    def __init__(self, bucket, org, precision=DEFAULT_WRITE_PRECISION) -> None:
+    def __init__(self, bucket, org, precision=DEFAULT_WRITE_PRECISION, **kwargs) -> None:
         self.bucket = bucket
         self.org = org
         self.precision = precision
+        self.kwargs = kwargs
         pass
 
     def __hash__(self) -> int:
@@ -178,8 +179,8 @@ def __eq__(self, o: object) -> bool:
             and self.bucket == o.bucket and self.org == o.org and self.precision == o.precision
 
     def __str__(self) -> str:
-        return '_BatchItemKey[bucket:\'{}\', org:\'{}\', precision:\'{}\']' \
-            .format(str(self.bucket), str(self.org), str(self.precision))
+        return '_BatchItemKey[bucket:\'{}\', org:\'{}\', precision:\'{}\', kwargs: \'{}\']' \
+            .format(str(self.bucket), str(self.org), str(self.precision), str(self.kwargs))
 
 
 class _BatchItem(object):
@@ -378,8 +379,6 @@ def write(self, bucket: str, org: str = None,
                 data_frame.index = pd.to_datetime(data_frame.index, unit='s')
 
         """  # noqa: E501
-        print("DEBUG WriteApi.write")
-        print(f"DEBUG kwargs {kwargs}")
         org = get_org_query_param(org=org, client=self._influxdb_client)
 
         self._append_default_tags(record)
@@ -477,7 +476,7 @@ def _write_batching(self, bucket, org, data,
             precision = self._write_options.write_precision
 
         if isinstance(data, bytes):
-            _key = _BatchItemKey(bucket, org, precision)
+            _key = _BatchItemKey(bucket, org, precision, **kwargs)
             self._subject.on_next(_BatchItem(key=_key, data=data))
 
         elif isinstance(data, str):
@@ -527,9 +526,8 @@ def _write_batching(self, bucket, org, data,
 
         return None
 
-    def _http(self, batch_item: _BatchItem):
+    def _http(self, batch_item: _BatchItem, **kwargs):
         logger.debug("Write time series data into InfluxDB: %s", batch_item)
-        print("DEBUG _http")
 
         if self._retry_callback:
             def _retry_callback_delegate(exception):
@@ -542,29 +540,28 @@ def _retry_callback_delegate(exception):
         retry = self._write_options.to_retry_strategy(retry_callback=_retry_callback_delegate)
 
         self._post_write(False, batch_item.key.bucket, batch_item.key.org, batch_item.data,
-                         batch_item.key.precision, no_sync, urlopen_kw={'retries': retry})
+                         batch_item.key.precision, no_sync, urlopen_kw={'retries': retry}, **kwargs)
 
         logger.debug("Write request finished %s", batch_item)
 
         return _BatchResponse(data=batch_item)
 
     def _post_write(self, _async_req, bucket, org, body, precision, no_sync, **kwargs):
-        print("DEBUG WriteApi._post_write")
-        print(f"DEBUG kwargs {kwargs} ")
-        return self._write_service.post_write(org=org, bucket=bucket, body=body, precision=precision,
+            return self._write_service.post_write(org=org, bucket=bucket, body=body, precision=precision,
                                               no_sync=no_sync,
                                               async_req=_async_req,
                                               content_type="text/plain; charset=utf-8",
                                               **kwargs)
 
+
     def _to_response(self, data: _BatchItem, delay: timedelta):
 
         return rx.of(data).pipe(
             ops.subscribe_on(self._write_options.write_scheduler),
             # use delay if its specified
             ops.delay(duetime=delay, scheduler=self._write_options.write_scheduler),
             # invoke http call
-            ops.map(lambda x: self._http(x)),
+            ops.map(lambda x: self._http(x, **x.key.kwargs)),
             # catch exception to fail batch response
             ops.catch(handler=lambda exception, source: rx.just(_BatchResponse(exception=exception, data=data))),
         )

influxdb_client_3/write_client/service/write_service.py

@@ -42,9 +42,6 @@ def post_write(self, org, bucket, body, **kwargs):  # noqa: E501,D401,D403
                  returns the request thread.
         """  # noqa: E501
 
-        print("DEBUG WriteService.post_write")
-        print(f"DEBUG kwargs {kwargs}")
-
         kwargs['_return_http_data_only'] = True
         if kwargs.get('async_req'):
             thread = self.post_write_with_http_info(org, bucket, body, **kwargs)  # noqa: E501
@@ -139,8 +136,6 @@ def post_write_with_http_info(self, org, bucket, body, **kwargs):  # noqa: E501,
                  If the method is called asynchronously,
                  returns the request thread.
         """  # noqa: E501
-        print("DEBUG WriteService.post_write_with_http_info()")
-        print(f"DEBUG kwargs {kwargs}")
         # noqa: E501
         local_var_params, path, path_params, query_params, header_params, body_params = \
             self._post_write_prepare(org, bucket, body, **kwargs)  # noqa: E501

tests/test_influxdb_client_3_integration.py

@@ -317,11 +317,12 @@ def set_error_result(rt, rd, rx):
             host=self.host,
             database=self.database,
             token=self.token,
+            write_timeout=20,
             write_client_options=write_client_options(
                 error_callback=set_error_result,
                 write_options=WriteOptions(
                     max_retry_time=0,  # disable retries
-                    timeout=20,
+                    # timeout=20,
                     write_type=WriteType.batching,
                     max_retries=0,
                     batch_size=1,
@@ -363,11 +364,12 @@ def retry_cb(args, data, excp):
             write_client_options=write_client_options(
                 error_callback=set_error_result,
                 retry_callback=retry_cb,
+                write_timeout=20,
                 write_options=WriteOptions(
                     max_retry_time=10000,
                     max_retry_delay=100,
                     retry_interval=100,
-                    timeout=20,
+                    # timeout=20,
                     max_retries=3,
                     batch_size=1,
                 )
@@ -387,32 +389,6 @@ def retry_cb(args, data, excp):
         self.assertIsInstance(ErrorResult["rx"], MaxRetryError)
         self.assertIsInstance(ErrorResult["rx"].reason, Url3TimeoutError)
 
-    def test_write_timeout_argument(self):
-        with self.assertRaises(Url3TimeoutError):
-            localClient = InfluxDBClient3(
-                host=self.host,
-                database=self.database,
-                token=self.token,
-                write_client_options=write_client_options(
-                    # error_callback=set_error_result,
-                    # retry_callback=retry_cb,
-                    # write_options=WriteOptions(
-                    #    max_retry_time=10000,
-                    #    max_retry_delay=0,
-                    #    retry_interval=0,
-                    #    timeout=20,
-                    #    max_retries=0,
-                    #    batch_size=1,
-                    #  )
-                )
-            )
-
-            print(f"DEBUG client.write_options {localClient._write_client_options['write_options'].__dict__}")
-
-            lp = "test_write_timeout,location=harfa fVal=3.14,iVal=42i"
-            # TODO investigate how the kwarg below can work with WriteType.batching...
-            localClient.write(lp, _request_timeout=1)
-
     @pytest.mark.skip(reason="flaky in CircleCI - server often responds in less than 1 millisecond.")
     def test_query_timeout(self):
         localClient = InfluxDBClient3(
@@ -435,3 +411,43 @@ def test_query_timeout_per_call_override(self):
 
         with self.assertRaisesRegex(InfluxDB3ClientQueryError, ".*Deadline Exceeded.*"):
             localClient.query("SELECT * FROM data", timeout=0.0001)
+
+    def test_write_timeout_per_call_override(self):
+
+        ErrorResult = {"rt": None, "rd": None, "rx": None}
+
+        def set_error_result(rt, rd, rx):
+            nonlocal ErrorResult
+            ErrorResult = {"rt": rt, "rd": rd, "rx": rx}
+
+        retry_ct = 0
+
+        def retry_cb(args, data, excp):
+            nonlocal retry_ct
+            retry_ct += 1
+            if excp is not None:
+                raise excp
+
+        localClient = InfluxDBClient3(
+            host=self.host,
+            token=self.token,
+            database=self.database,
+            # write_timeout=3000,
+            write_client_options=write_client_options(
+                error_callback=set_error_result,
+                retry_callback=retry_cb,
+                write_options=WriteOptions(
+                    batch_size=1,
+                ),
+
+            )
+        )
+
+        lp = "test_write_timeout,location=harfa fVal=3.14,iVal=42i"
+        # with self.assertRaises(Url3TimeoutError):
+        localClient.write(lp, _request_timeout=1)
+
+        # wait for batcher attempt last write retry
+        time.sleep(0.1)
+
+        print(f"DEBUG ErrorResult {ErrorResult}, retry_ct {retry_ct}")

tests/test_write_local_server.py

@@ -23,7 +23,7 @@ def assert_request_made(httpserver, matcher):
         httpserver.check_assertions()
 
     @staticmethod
-    def delay_response(httpserver, delay=1.0):
+    def delay_response(httpserver: HTTPServer, delay=1.0):
         httpserver.expect_request(re.compile(".*")).respond_with_handler(lambda request: time.sleep(delay))
 
     def test_write_default_params(self, httpserver: HTTPServer):
@@ -169,7 +169,7 @@ def test_write_with_write_timeout(self, httpserver: HTTPServer):
             ).write(self.SAMPLE_RECORD)
 
     def test_write_with_timeout_arg(self, httpserver: HTTPServer):
-        self.set_response_status(httpserver, 200)
+        self.delay_response(httpserver, 0.5)
 
         with pytest.raises(urllib3_TimeoutError):
             InfluxDBClient3(