refactor: multiprocess helper class

Author: NguyenHoangSon96

CHANGELOG.md

@@ -2,6 +2,65 @@
 
 ## 0.21.0 [unreleased]
 
+### Breaking Changes
+
+1. [#217](https://github.com/InfluxCommunity/influxdb3-python/pull/217): Make Write API simpler and consistent with other v3 clients.
+    - Remove unused `InfluxLoggingHandler` class.
+    - WriteApi now has all functions to write to InfluxDB. `Configuration`, `ApiClient`, `WriteService`, `InfluxDBClient` are no longer needed.
+    All settings needed for writing now can be passed to the constructor of WriteApi.
+    ``` python
+        import time
+        from influxdb_client_3.write_client import WriteApi
+        from influxdb_client_3.write_client._sync import rest_client
+        from influxdb_client_3 import WriteOptions, write_client_options
+
+        default_header = {
+            'Authorization': 'Token my-token'
+        }
+        rest = rest_client.RestClient(
+            base_url='http://localhost:8181',
+            default_header=default_header,
+            verify_ssl=True,
+            ssl_ca_cert=None,
+            cert_file=None,
+            cert_key_file=None,
+            cert_key_password=None,
+            ssl_context=None,
+            proxy=None,
+            proxy_headers=None,
+            retries=False,
+            debug=False,
+            connection_pool_maxsize=25
+        )
+        
+        wco=write_client_options(write_options=WriteOptions(batch_size=100)))
+        write_api = WriteApi(
+            bucket='bucket_name',
+            org='my-org',
+            default_header=default_header,
+            gzip_threshold=None,
+            enable_gzip=False,
+            auth_scheme='Token',
+            timeout=None,
+            rest_client=rest,
+            point_settings=None,
+            **wco
+        )
+
+        test_id = time.time_ns()
+        write_api.write(record=f"cpu,type=used ram=16,test_id={test_id}i")
+        write_api.close()
+
+    ``` 
+    - rest_client.RestClient will now be responsible for low-level handling the HTTP requests. 
+    - `InfluxDBClient3` constructor will have additional parameters for configuring the RestClient and WriteApi.
+      - auth_scheme,
+      - enable_gzip,
+      - gzip_threshold,   
+      - point_settings,
+      - debug,   
+    - Refactor Multiprocessing helper class.
+
 ## 0.20.0 [2026-06-11]
 
 ### Features

influxdb_client_3/__init__.py

@@ -1,10 +1,16 @@
+import multiprocessing
+
 import importlib.util
+import json
 import os
 import urllib.parse
 from typing import Any, List, Literal, Optional, TYPE_CHECKING
 
 import pyarrow as pa
 
+from influxdb_client_3.version import USER_AGENT
+from influxdb_client_3.write_client._sync import rest_client as rest
+
 if TYPE_CHECKING:
     import pandas as pd
     import polars as pl
@@ -14,7 +20,7 @@
 from influxdb_client_3.exceptions import InfluxDBError
 from influxdb_client_3.query.query_api import QueryApi as _QueryApi, QueryApiOptionsBuilder
 from influxdb_client_3.read_file import UploadFile
-from influxdb_client_3.write_client import InfluxDBClient as _InfluxDBClient, WriteOptions, Point
+from influxdb_client_3.write_client import WriteOptions, Point
 from influxdb_client_3.write_client.client.write_api import WriteApi as _WriteApi, SYNCHRONOUS, ASYNCHRONOUS, \
     PointSettings, DefaultWriteOptions, WriteType
 from influxdb_client_3.write_client.domain.write_precision import WritePrecision
@@ -189,11 +195,16 @@ def __init__(
             org=None,
             database=None,
             token=None,
+            auth_scheme=None,
+            enable_gzip=False,
+            gzip_threshold=None,
             write_client_options=None,
             flight_client_options=None,
             write_port_overwrite=None,
             query_port_overwrite=None,
             disable_grpc_compression=False,
+            point_settings=None,
+            debug=False,
             **kwargs):
         """
         Initialize an InfluxDB client.
@@ -212,6 +223,14 @@ def __init__(
         :type flight_client_options: dict[str, any]
         :param disable_grpc_compression: Disable gRPC compression for Flight query responses. Default is False.
         :type disable_grpc_compression: bool
+        :param point_settings The settings for Points
+        :type point_settings: PointSettings
+        :param debug: enable verbose logging of http requests
+        :type debug: bool
+        :param enable_gzip: Enable GZIP compression for write requests.
+        :type enable_gzip: bool
+        :param gzip_threshold: Minimum payload size (bytes) to trigger GZIP when enable_gzip is True.
+        :type gzip_threshold: int
         :key auth_scheme: token authentication scheme. Set to "Bearer" for Edge.
         :key bool verify_ssl: Set this to false to skip verifying SSL certificate when calling API from https server.
         :key str ssl_ca_cert: Set this to customize the certificate file to verify the peer.
@@ -235,6 +254,10 @@ def __init__(
         :key bool write_no_sync: disable sync confirmation on V3 API endpoint writes.
         :key list[str] profilers: list of enabled Flux profilers
         """
+        for key in ["host", "token", "database"]:
+            if locals().get(key) is None:
+                raise Exception(f"The '{key}' key is required")
+
         self._org = org if org is not None else "default"
         self._database = database
         self._token = token
@@ -293,14 +316,49 @@ def __init__(
         if write_port_overwrite is not None:
             port = write_port_overwrite
 
-        self._client = _InfluxDBClient(
-            url=f"{scheme}://{hostname}:{port}",
-            token=self._token,
+        auth_schema = 'Token' if auth_scheme is None else auth_scheme
+        default_header = {
+            'User-Agent': USER_AGENT
+        }
+        if self._token is not None:
+            default_header['Authorization'] = f'{auth_schema} {self._token}'
+        self.base_url = f"{scheme}://{hostname}:{port}"
+        self.default_header = default_header
+        self.rest_client = rest.RestClient(
+            base_url=self.base_url,
+            default_header=default_header,
+            verify_ssl=kwargs.get('verify_ssl', True),
+            ssl_ca_cert=kwargs.get('ssl_ca_cert', None),
+            cert_file=kwargs.get('cert_file', None),
+            cert_key_file=kwargs.get('cert_key_file', None),
+            cert_key_password=kwargs.get('cert_key_password', None),
+            ssl_context=kwargs.get('ssl_context', None),
+            proxy=kwargs.get('proxy', None),
+            proxy_headers=kwargs.get('proxy_headers', None),
+            retries=kwargs.get('retries', False),
+            debug=debug,
+            connection_pool_maxsize=kwargs.get('connection_pool_maxsize', multiprocessing.cpu_count() * 5,)
+        )
+
+        if point_settings is None:
+            point_settings = PointSettings()
+
+        # Keep WriteOptions.timeout in sync with the resolved write_timeout
+        if isinstance(self._write_client_options, dict) and self._write_client_options.get("write_options") is not None:
+            self._write_client_options["write_options"].timeout = write_timeout
+
+        self._write_api = _WriteApi(
+            bucket=self._database,
             org=self._org,
+            gzip_threshold=gzip_threshold,
+            enable_gzip=enable_gzip,
+            auth_scheme=auth_scheme,
             timeout=write_timeout,
-            **kwargs)
-
-        self._write_api = _WriteApi(influxdb_client=self._client, **self._write_client_options)
+            default_header=default_header,
+            rest_client=self.rest_client,
+            point_settings=point_settings,
+            **self._write_client_options
+        )
 
         if query_port_overwrite is not None:
             port = query_port_overwrite
@@ -658,32 +716,31 @@ async def query_async(self, query: str, language: str = "sql", mode: str = "all"
         except ArrowException as e:
             raise InfluxDB3ClientQueryError(f"Error while executing query: {e}")
 
-    def get_server_version(self) -> str:
+    def get_server_version(self) -> Optional[str]:
         """
-        Get the version of the connected InfluxDB server.
+        Retrieves the server version by querying the designated endpoint and
+        extracting the version information from either response headers or
+        the response body.
 
-        This method makes a ping request to the server and extracts the version information
-        from either the response headers or response body.
+        This method interacts with a REST API endpoint to fetch the server's
+        version details, which might be stored in a specific HTTP header or
+        available in the response body as part of a JSON structure.
 
-        :return: The version string of the InfluxDB server.
-        :rtype: str
+        :return: The version string of the server if available, otherwise None.
+        :rtype: Optional[str]
         """
-        version = None
-        (resp_body, _, header) = self._client.api_client.call_api(
-            resource_path="/ping",
-            method="GET",
-            response_type=object
-        )
-
-        for key, value in header.items():
+        resp = self.rest_client.request(path='/ping', method="GET", headers=self.default_header)
+        for key, value in resp.getheaders().items():
             if key.lower() == "x-influxdb-version":
-                version = value
-                break
+                return value
 
-        if version is None and isinstance(resp_body, dict):
-            version = resp_body['version']
-
-        return version
+        try:
+            if resp.data is not None:
+                return json.loads(resp.data).get("version")
+            else:
+                return None
+        except (ValueError, TypeError):
+            return None
 
     def flush(self):
         """
@@ -700,9 +757,12 @@ def flush(self):
 
     def close(self):
         """Close the client and clean up resources."""
-        self._write_api.close()
-        self._query_api.close()
-        self._client.close()
+        if self._write_api is not None:
+            self._write_api.close()
+        if self._query_api is not None:
+            self._query_api.close()
+        if self.rest_client is not None:
+            self.rest_client.close()
 
     def __enter__(self):
         return self

influxdb_client_3/write_client/__init__.py

@@ -4,15 +4,9 @@
 
 from __future__ import absolute_import
 
-from influxdb_client_3.write_client.client.write_api import WriteApi, WriteOptions
-from influxdb_client_3.write_client.client.influxdb_client import InfluxDBClient
-from influxdb_client_3.write_client.client.logging_handler import InfluxLoggingHandler
+from influxdb_client_3.version import VERSION
 from influxdb_client_3.write_client.client.write.point import Point
-
-from influxdb_client_3.write_client.service.write_service import WriteService
-
+from influxdb_client_3.write_client.client.write_api import WriteApi, WriteOptions
 from influxdb_client_3.write_client.domain.write_precision import WritePrecision
 
-from influxdb_client_3.write_client.configuration import Configuration
-from influxdb_client_3.version import VERSION
 __version__ = VERSION