test: add multiprocess helper test

Author: NguyenHoangSon96

influxdb_client_3/write_client/_sync/rest_client.py

@@ -167,6 +167,8 @@ def request(self, method, url, query_params=None, headers=None,
             r.data = ''
 
         if not 200 <= r.status <= 299:
+            print('hahaha')
+            print(r.data)
             raise ApiException(http_resp=r)
 
         return r

influxdb_client_3/write_client/client/util/multiprocessing_helper.py

@@ -7,6 +7,7 @@
 import logging
 import multiprocessing
 
+from influxdb_client_3 import InfluxDBClient3, write_client_options
 from influxdb_client_3.write_client import WriteOptions
 from influxdb_client_3.exceptions import InfluxDBError
 
@@ -148,11 +149,22 @@ def write(self, **kwargs) -> None:
     def run(self):
         """Initialize ``InfluxDBClient`` and waits for data to writes into InfluxDB."""
         # Initialize Client and Write API
-        self.client = InfluxDBClient(**self.kwargs)
-        self.write_api = self.client.write_api(write_options=self.kwargs.get('write_options', WriteOptions()),
-                                               success_callback=self.kwargs.get('success_callback', _success_callback),
-                                               error_callback=self.kwargs.get('error_callback', _error_callback),
-                                               retry_callback=self.kwargs.get('retry_callback', _retry_callback))
+        wco = write_client_options(write_options=self.kwargs.get('write_options', WriteOptions()),
+                                   success_callback=self.kwargs.get('success_callback', _success_callback),
+                                   error_callback=self.kwargs.get('error_callback', _error_callback),
+                                   retry_callback=self.kwargs.get('retry_callback', _retry_callback)
+                                   )
+
+        # Still need to create the InfluxDBClient3 because the init logics of InfluxDBClient3 will create the WriteApi.
+        # it will make WriteApi class created properly.
+        self.client = InfluxDBClient3(write_client_options=wco, **self.kwargs)
+
+        # Close and set _query_api to None because query_api is not needed in this process.
+        # We only need write_api.
+        self.client._query_api.close()
+        self.client._query_api = None
+
+        self.write_api = self.client._write_api
         # Infinite loop - until poison pill
         while True:
             next_record = self.queue_.get()
@@ -181,7 +193,7 @@ def terminate(self) -> None:
             self.write_api.__del__()
             self.write_api = None
         if self.client:
-            self.client.__del__()
+            self.client.close()
             self.client = None
             logger.info("closed")
 

influxdb_client_3/write_client/client/write_api.py

@@ -965,6 +965,49 @@ def _should_gzip(self, payload: str, enable_gzip: bool = False, gzip_threshold:
 
         return False
 
+    @staticmethod
+    def _on_error(ex):
+        logger.error("unexpected error during batching: %s", ex)
+
+    def _to_response(self, data: _BatchItem, delay: datetime.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, **x.key.kwargs)),
+            # catch exception to fail batch response
+            ops.catch(handler=lambda exception, source: rx.just(_BatchResponse(exception=exception, data=data))),
+        )
+
+    def _on_next(self, response: _BatchResponse):
+        if response.exception:
+            logger.error("The batch item wasn't processed successfully because: %s", response.exception)
+            if self._error_callback:
+                try:
+                    self._error_callback(response.data.to_key_tuple(), response.data.data, response.exception)
+                except Exception as e:
+                    """
+                    Unfortunately, because callbacks are user-provided generic code, exceptions can be entirely
+                    arbitrary
+
+                    We trap it, log that it occurred and then proceed - there's not much more that we can
+                    really do.
+                    """
+                    logger.error("The configured error callback threw an exception: %s", e)
+
+        else:
+            logger.debug("The batch item: %s was processed successfully.", response)
+            if self._success_callback:
+                try:
+                    self._success_callback(response.data.to_key_tuple(), response.data.data)
+                except Exception as e:
+                    logger.error("The configured success callback threw an exception: %s", e)
+
+    def _on_complete(self):
+        self._disposable.dispose()
+        logger.debug("the batching processor was disposed")
+
     def _append_default_tag(self, key, val, record):
         from influxdb_client_3.write_client import Point
         if isinstance(record, bytes) or isinstance(record, str):
@@ -1049,49 +1092,6 @@ def __del__(self):
         """Close WriteApi."""
         self.close()
 
-    @staticmethod
-    def _on_error(ex):
-        logger.error("unexpected error during batching: %s", ex)
-
-    def _on_complete(self):
-        self._disposable.dispose()
-        logger.debug("the batching processor was disposed")
-
-    def _to_response(self, data: _BatchItem, delay: datetime.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, **x.key.kwargs)),
-            # catch exception to fail batch response
-            ops.catch(handler=lambda exception, source: rx.just(_BatchResponse(exception=exception, data=data))),
-        )
-
-    def _on_next(self, response: _BatchResponse):
-        if response.exception:
-            logger.error("The batch item wasn't processed successfully because: %s", response.exception)
-            if self._error_callback:
-                try:
-                    self._error_callback(response.data.to_key_tuple(), response.data.data, response.exception)
-                except Exception as e:
-                    """
-                    Unfortunately, because callbacks are user-provided generic code, exceptions can be entirely
-                    arbitrary
-
-                    We trap it, log that it occurred and then proceed - there's not much more that we can
-                    really do.
-                    """
-                    logger.error("The configured error callback threw an exception: %s", e)
-
-        else:
-            logger.debug("The batch item: %s was processed successfully.", response)
-            if self._success_callback:
-                try:
-                    self._success_callback(response.data.to_key_tuple(), response.data.data)
-                except Exception as e:
-                    logger.error("The configured success callback threw an exception: %s", e)
-
     def __setstate__(self, state):
         """Set your object with the provided dict."""
         self.__dict__.update(state)

tests/test_influxdb_client_3_integration.py

@@ -14,6 +14,7 @@
 from influxdb_client_3 import InfluxDBClient3, write_client_options, WriteOptions, \
     WriteType, InfluxDB3ClientQueryError
 from influxdb_client_3.exceptions import InfluxDBError, InfluxDBPartialWriteError
+from influxdb_client_3.write_client.client.util.multiprocessing_helper import MultiprocessingWriter
 from influxdb_client_3.write_client.rest import ApiException
 from tests.util import asyncio_run, lp_to_py_object