feat: fast no-sync write support (#142)

Author: jansimonb

CHANGELOG.md

@@ -2,6 +2,17 @@
 
 ## 0.14.0 [unreleased]
 
+### Features
+
+1. [#142](https://github.com/InfluxCommunity/influxdb3-python/pull/142):  Support fast writes without waiting for WAL
+   persistence:
+   - New write option (`WriteOptions.no_sync`) added: `True` value means faster write but without the confirmation that
+     the data was persisted. Default value: `False`.
+   - **Supported by self-managed InfluxDB 3 Core and Enterprise servers only!**
+   - Also configurable via environment variable (`INFLUX_WRITE_NO_SYNC`).
+   - Long precision string values added from v3 HTTP API: `"nanosecond"`, `"microsecond"`, `"millisecond"`,
+     `"second"` (     in addition to the existing `"ns"`, `"us"`, `"ms"`, `"s"`).
+
 ## 0.13.0 [2025-05-20]
 
 ### Features

influxdb_client_3/__init__.py

@@ -7,10 +7,10 @@
 from pyarrow import ArrowException
 
 from influxdb_client_3.exceptions import InfluxDB3ClientQueryError
+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.exceptions import InfluxDBError
 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
@@ -24,6 +24,7 @@
 INFLUX_PRECISION = "INFLUX_PRECISION"
 INFLUX_AUTH_SCHEME = "INFLUX_AUTH_SCHEME"
 INFLUX_GZIP_THRESHOLD = "INFLUX_GZIP_THRESHOLD"
+INFLUX_WRITE_NO_SYNC = "INFLUX_WRITE_NO_SYNC"
 
 
 def write_client_options(**kwargs):
@@ -112,9 +113,15 @@ def _parse_precision(precision):
     :rtype: WritePrecision
     :raises ValueError: If the provided precision is not valid.
     """
-    if precision not in [WritePrecision.NS, WritePrecision.MS, WritePrecision.S, WritePrecision.US]:
-        raise ValueError(f"Invalid precision value: {precision}")
-    return precision
+    if precision == WritePrecision.NS or precision == "nanosecond":
+        return WritePrecision.NS
+    if precision == WritePrecision.US or precision == "microsecond":
+        return WritePrecision.US
+    if precision == WritePrecision.MS or precision == "millisecond":
+        return WritePrecision.MS
+    if precision == WritePrecision.S or precision == "second":
+        return WritePrecision.S
+    raise ValueError(f"Invalid precision value: {precision}")
 
 
 def _parse_gzip_threshold(threshold):
@@ -141,6 +148,21 @@ def _parse_gzip_threshold(threshold):
     return threshold
 
 
+def _parse_write_no_sync(write_no_sync):
+    """
+    Parses and validates the provided write no sync value.
+
+    This function ensures that the given value is a valid boolean,
+    and it raises an appropriate error if the value is not valid.
+
+    :param write_no_sync: The input value to be parsed and validated.
+    :type write_no_sync: Any
+    :return: The validated write no sync value as an boolean.
+    :rtype: bool
+    """
+    return write_no_sync.strip().lower() in ['true', '1', 't', 'y', 'yes']
+
+
 class InfluxDBClient3:
     def __init__(
             self,
@@ -197,14 +219,17 @@ def __init__(
 
         write_type = DefaultWriteOptions.write_type.value
         write_precision = DefaultWriteOptions.write_precision.value
+        write_no_sync = DefaultWriteOptions.no_sync.value
         if isinstance(write_client_options, dict) and write_client_options.get('write_options') is not None:
             write_opts = write_client_options['write_options']
             write_type = getattr(write_opts, 'write_type', write_type)
             write_precision = getattr(write_opts, 'write_precision', write_precision)
+            write_no_sync = getattr(write_opts, 'no_sync', write_no_sync)
 
         write_options = WriteOptions(
             write_type=write_type,
             write_precision=write_precision,
+            no_sync=write_no_sync,
         )
 
         self._write_client_options = {
@@ -286,6 +311,10 @@ def from_env(cls, **kwargs: Any) -> 'InfluxDBClient3':
             kwargs['gzip_threshold'] = _parse_gzip_threshold(gzip_threshold)
             kwargs['enable_gzip'] = True
 
+        write_no_sync = os.getenv(INFLUX_WRITE_NO_SYNC)
+        if write_no_sync is not None:
+            write_options.no_sync = _parse_write_no_sync(write_no_sync)
+
         precision = os.getenv(INFLUX_PRECISION)
         if precision is not None:
             write_options.write_precision = _parse_precision(precision)

influxdb_client_3/write_client/client/_base.py

@@ -282,7 +282,7 @@ def update_request_header_params(self, path: str, params: dict, should_gzip: boo
         super().update_request_header_params(path, params, should_gzip)
         if should_gzip:
             # GZIP Request
-            if path == '/api/v2/write':
+            if path == '/api/v2/write' or path == '/api/v3/write_lp':
                 params["Content-Encoding"] = "gzip"
                 params["Accept-Encoding"] = "identity"
                 pass
@@ -298,7 +298,7 @@ def update_request_body(self, path: str, body, should_gzip: bool = False):
         _body = super().update_request_body(path, body, should_gzip)
         if should_gzip:
             # GZIP Request
-            if path == '/api/v2/write':
+            if path == '/api/v2/write' or path == '/api/v3/write_lp':
                 import gzip
                 if isinstance(_body, bytes):
                     return gzip.compress(data=_body)

influxdb_client_3/write_client/client/write_api.py

@@ -24,6 +24,8 @@
 from influxdb_client_3.write_client.domain import WritePrecision
 from influxdb_client_3.write_client.rest import _UTF_8_encoding
 
+DEFAULT_WRITE_NO_SYNC = False
+
 logger = logging.getLogger('influxdb_client_3.write_client.client.write_api')
 
 if _HAS_DATACLASS:
@@ -41,7 +43,8 @@ class WriteType(Enum):
 
 class DefaultWriteOptions(Enum):
     write_type = WriteType.synchronous
-    write_precision = WritePrecision.NS
+    write_precision = DEFAULT_WRITE_PRECISION
+    no_sync = DEFAULT_WRITE_NO_SYNC
 
 
 class WriteOptions(object):
@@ -57,6 +60,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
                  exponential_base=2,
                  max_close_wait=300_000,
                  write_precision=DEFAULT_WRITE_PRECISION,
+                 no_sync=DEFAULT_WRITE_NO_SYNC,
                  write_scheduler=ThreadPoolScheduler(max_workers=1)) -> None:
         """
         Create write api configuration.
@@ -72,7 +76,9 @@ def __init__(self, write_type: WriteType = WriteType.batching,
         :param max_retry_delay: the maximum delay between each retry attempt in milliseconds
         :param max_retry_time: total timeout for all retry attempts in milliseconds, if 0 retry is disabled
         :param exponential_base: base for the exponential retry delay
-        :parama max_close_wait: the maximum time to wait for writes to be flushed if close() is called
+        :param max_close_wait: the maximum time to wait for writes to be flushed if close() is called
+        :param write_precision: precision to use when writing points to InfluxDB
+        :param no_sync: skip waiting for WAL persistence on write
         :param write_scheduler:
         """
         self.write_type = write_type
@@ -87,6 +93,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
         self.write_scheduler = write_scheduler
         self.max_close_wait = max_close_wait
         self.write_precision = write_precision
+        self.no_sync = no_sync
 
     def to_retry_strategy(self, **kwargs):
         """
@@ -202,7 +209,7 @@ def _body_reduce(batch_items):
 
 class WriteApi(_BaseWriteApi):
     """
-    Implementation for '/api/v2/write' endpoint.
+    Implementation for '/api/v2/write' and '/api/v3/write_lp' endpoint.
 
     Example:
         .. code-block:: python
@@ -375,14 +382,16 @@ def write(self, bucket: str, org: str = None,
             return self._write_batching(bucket, org, record,
                                         write_precision, **kwargs)
 
+        no_sync = self._write_options.no_sync
+
         payloads = defaultdict(list)
         self._serialize(record, write_precision, payloads, **kwargs)
 
         _async_req = True if self._write_options.write_type == WriteType.asynchronous else False
 
         def write_payload(payload):
             final_string = b'\n'.join(payload[1])
-            return self._post_write(_async_req, bucket, org, final_string, payload[0])
+            return self._post_write(_async_req, bucket, org, final_string, payload[0], no_sync)
 
         results = list(map(write_payload, payloads.items()))
         if not _async_req:
@@ -519,18 +528,21 @@ def _retry_callback_delegate(exception):
         else:
             _retry_callback_delegate = None
 
+        no_sync = self._write_options.no_sync
+
         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, urlopen_kw={'retries': retry})
+                         batch_item.key.precision, no_sync, urlopen_kw={'retries': retry})
 
         logger.debug("Write request finished %s", batch_item)
 
         return _BatchResponse(data=batch_item)
 
-    def _post_write(self, _async_req, bucket, org, body, precision, **kwargs):
+    def _post_write(self, _async_req, bucket, org, body, precision, no_sync, **kwargs):
 
         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)

influxdb_client_3/write_client/domain/write_precision_converter.py

@@ -0,0 +1,30 @@
+from influxdb_client_3.write_client.domain import WritePrecision
+
+
+class WritePrecisionConverter(object):
+
+    @staticmethod
+    def to_v2_api_string(precision):
+        """
+        Converts WritePrecision to its string representation for V2 API.
+        """
+        if precision in [WritePrecision.NS, WritePrecision.US, WritePrecision.MS, WritePrecision.S]:
+            return precision
+        else:
+            raise ValueError("Unsupported precision '%s'" % precision)
+
+    @staticmethod
+    def to_v3_api_string(precision):
+        """
+        Converts WritePrecision to its string representation for V3 API.
+        """
+        if precision == WritePrecision.NS:
+            return "nanosecond"
+        elif precision == WritePrecision.US:
+            return "microsecond"
+        elif precision == WritePrecision.MS:
+            return "millisecond"
+        elif precision == WritePrecision.S:
+            return "second"
+        else:
+            raise ValueError("Unsupported precision '%s'" % precision)

influxdb_client_3/write_client/service/signin_service.py

@@ -106,7 +106,7 @@ async def post_signin_async(self, **kwargs):  # noqa: E501,D401,D403
             urlopen_kw=kwargs.get('urlopen_kw', None))
 
     def _post_signin_prepare(self, **kwargs):  # noqa: E501,D401,D403
-        local_var_params = locals()
+        local_var_params = dict(locals())
 
         all_params = ['zap_trace_span', 'authorization']  # noqa: E501
         self._check_operation_params('post_signin', all_params, local_var_params)

influxdb_client_3/write_client/service/signout_service.py

@@ -103,7 +103,7 @@ async def post_signout_async(self, **kwargs):  # noqa: E501,D401,D403
             urlopen_kw=kwargs.get('urlopen_kw', None))
 
     def _post_signout_prepare(self, **kwargs):  # noqa: E501,D401,D403
-        local_var_params = locals()
+        local_var_params = dict(locals())
 
         all_params = ['zap_trace_span']  # noqa: E501
         self._check_operation_params('post_signout', all_params, local_var_params)