fix: resolve write option kwargs before post_write call

Author: alespour

influxdb_client_3/write_client/client/write_api.py

@@ -320,6 +320,14 @@ def __init__(self,
             # TODO above message has link to Influxdb2 API __NOT__ Influxdb3 API !!! - illustrates different API
             warnings.warn(message, DeprecationWarning)
 
+    def _resolve_write_request_options(self, kwargs):
+        no_sync = kwargs.pop('no_sync', self._write_options.no_sync)
+        accept_partial = kwargs.pop('accept_partial', self._write_options.accept_partial)
+        use_v2_api = kwargs.pop('use_v2_api', self._write_options.use_v2_api)
+        if use_v2_api and no_sync:
+            raise ValueError("invalid write options: no_sync cannot be used with use_v2_api")
+        return no_sync, accept_partial, use_v2_api
+
     def write(self, bucket: str, org: str = None,
               record: Union[
                   str, Iterable['str'], Point, Iterable['Point'], dict, Iterable['dict'], bytes, Iterable['bytes'],
@@ -400,20 +408,21 @@ def write(self, bucket: str, org: str = None,
             write_precision = self._write_options.write_precision
 
         self._write_options.validate()
+        kwargs = dict(kwargs)
+        no_sync, accept_partial, use_v2_api = self._resolve_write_request_options(kwargs)
 
         if 'tag_order' in kwargs:
             kwargs['tag_order'] = sanitize_tag_order(kwargs.get('tag_order'))
         else:
             kwargs['tag_order'] = self._write_options.tag_order
 
         if self._write_options.write_type is WriteType.batching:
+            kwargs['no_sync'] = no_sync
+            kwargs['accept_partial'] = accept_partial
+            kwargs['use_v2_api'] = use_v2_api
             return self._write_batching(bucket, org, record,
                                         write_precision, **kwargs)
 
-        no_sync = self._write_options.no_sync
-        accept_partial = self._write_options.accept_partial
-        use_v2_api = self._write_options.use_v2_api
-
         payloads = defaultdict(list)
         self._serialize(record, write_precision, payloads, **kwargs)
 
@@ -602,9 +611,8 @@ def _retry_callback_delegate(exception):
         else:
             _retry_callback_delegate = None
 
-        no_sync = self._write_options.no_sync
-        accept_partial = self._write_options.accept_partial
-        use_v2_api = self._write_options.use_v2_api
+        kwargs = dict(kwargs)
+        no_sync, accept_partial, use_v2_api = self._resolve_write_request_options(kwargs)
 
         retry = self._write_options.to_retry_strategy(retry_callback=_retry_callback_delegate)
 

tests/test_write_local_server.py

@@ -77,6 +77,20 @@ def test_write_with_no_sync_true(self, httpserver: HTTPServer):
             method="POST", uri="/api/v3/write_lp",
             query_string={"org": "ORG", "db": "DB", "precision": "microsecond", "no_sync": "true"}))
 
+    def test_write_with_no_sync_true_in_kwargs(self, httpserver: HTTPServer):
+        self.set_response_status(httpserver, 200)
+
+        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)
+            )
+        ).write(self.SAMPLE_RECORD, no_sync=True)
+
+        self.assert_request_made(httpserver, RequestMatcher(
+            method="POST", uri="/api/v3/write_lp",
+            query_string={"org": "ORG", "db": "DB", "precision": "microsecond", "no_sync": "true"}))
+
     def test_write_with_accept_partial_false(self, httpserver: HTTPServer):
         self.set_response_status(httpserver, 200)
 
@@ -113,6 +127,20 @@ def test_write_with_use_v2_api_true(self, httpserver: HTTPServer):
             method="POST", uri="/api/v2/write",
             query_string={"org": "ORG", "bucket": "DB", "precision": "us"}))
 
+    def test_write_with_use_v2_api_true_in_kwargs(self, httpserver: HTTPServer):
+        self.set_response_status(httpserver, 200)
+
+        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)
+            )
+        ).write(self.SAMPLE_RECORD, use_v2_api=True, accept_partial=False)
+
+        self.assert_request_made(httpserver, RequestMatcher(
+            method="POST", uri="/api/v2/write",
+            query_string={"org": "ORG", "bucket": "DB", "precision": "us"}))
+
     def test_write_with_v3_on_v2_server(self, httpserver: HTTPServer):
         self.set_response_status(httpserver, HTTPStatus.METHOD_NOT_ALLOWED)
 
@@ -186,6 +214,17 @@ def test_write_invalid_use_v2_api_and_no_sync(self, httpserver: HTTPServer):
         with pytest.raises(ValueError, match=r".*invalid write options: no_sync cannot be used with use_v2_api.*"):
             client.write(self.SAMPLE_RECORD)
 
+    def test_write_invalid_use_v2_api_and_no_sync_in_kwargs(self, httpserver: HTTPServer):
+        client = InfluxDBClient3(
+            host=(httpserver.url_for("/")), org="ORG", database="DB", token="TOKEN",
+            write_client_options=write_client_options(
+                write_options=WriteOptions(write_type=WriteType.synchronous)
+            )
+        )
+
+        with pytest.raises(ValueError, match=r".*invalid write options: no_sync cannot be used with use_v2_api.*"):
+            client.write(self.SAMPLE_RECORD, use_v2_api=True, no_sync=True)
+
     def test_write_with_timeout_in_write_options(self, httpserver: HTTPServer):
         self.delay_response(httpserver, 0.5)