chore: using _request_timeout write kwargs in write calls.

Author: karel-rehor

influxdb_client_3/__init__.py

@@ -377,6 +377,8 @@ 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

@@ -378,6 +378,8 @@ 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)
@@ -398,7 +400,7 @@ def write(self, bucket: str, org: str = None,
 
         def write_payload(payload):
             final_string = b'\n'.join(payload[1])
-            return self._post_write(_async_req, bucket, org, final_string, payload[0], no_sync)
+            return self._post_write(_async_req, bucket, org, final_string, payload[0], no_sync, **kwargs)
 
         results = list(map(write_payload, payloads.items()))
         if not _async_req:
@@ -527,6 +529,7 @@ def _write_batching(self, bucket, org, data,
 
     def _http(self, batch_item: _BatchItem):
         logger.debug("Write time series data into InfluxDB: %s", batch_item)
+        print("DEBUG _http")
 
         if self._retry_callback:
             def _retry_callback_delegate(exception):
@@ -546,6 +549,8 @@ def _retry_callback_delegate(exception):
         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,
                                               no_sync=no_sync,
                                               async_req=_async_req,

influxdb_client_3/write_client/service/write_service.py

@@ -41,10 +41,13 @@ def post_write(self, org, bucket, body, **kwargs):  # noqa: E501,D401,D403
                  If the method is called asynchronously,
                  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
-            # return self.post_write_with_http_info(org, bucket, body, **kwargs)  # noqa: E501
             return thread
         else:
             (data) = self.post_write_with_http_info(org, bucket, body, **kwargs)  # noqa: E501
@@ -136,7 +139,8 @@ 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("WriteService.post_write_with_http_info()")
+        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

@@ -387,6 +387,32 @@ 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(

tests/test_write_local_server.py

@@ -167,3 +167,19 @@ def test_write_with_write_timeout(self, httpserver: HTTPServer):
                 ),
                 enable_gzip=True
             ).write(self.SAMPLE_RECORD)
+
+    def test_write_with_timeout_arg(self, httpserver: HTTPServer):
+        self.set_response_status(httpserver, 200)
+
+        with pytest.raises(urllib3_TimeoutError):
+            InfluxDBClient3(
+                host=(httpserver.url_for("/")), org="ORG", database="DB", token="TOKEN",
+                write_client_options=write_client_options(
+                    write_options=WriteOptions(
+                        write_type=WriteType.synchronous,
+                        write_precision=WritePrecision.US,
+                        no_sync=True,
+                    )
+                ),
+                enable_gzip=True
+            ).write(self.SAMPLE_RECORD, _request_timeout=1)