feat: (WIP) timeouts - adds tests for write API timeout.

Author: karel-rehor

influxdb_client_3/__init__.py

@@ -352,7 +352,7 @@ def write(self, record=None, database=None, **kwargs):
             database = self._database
 
         try:
-            self._write_api.write(bucket=database, record=record, **kwargs)
+            return self._write_api.write(bucket=database, record=record, **kwargs)
         except InfluxDBError as e:
             raise e
 

influxdb_client_3/write_client/client/write_api.py

@@ -7,6 +7,7 @@
 from collections import defaultdict
 from datetime import timedelta
 from enum import Enum
+from multiprocessing.pool import ApplyResult
 from random import random
 from time import sleep
 from typing import Union, Any, Iterable, NamedTuple
@@ -264,6 +265,7 @@ def __init__(self,
         """
         super().__init__(influxdb_client=influxdb_client, point_settings=point_settings)
         self._write_options = write_options
+        # TODO - callbacks seem to be used with batching type only - could they be used with sync or async?
         self._success_callback = kwargs.get('success_callback', None)
         self._error_callback = kwargs.get('error_callback', None)
         self._retry_callback = kwargs.get('retry_callback', None)
@@ -303,6 +305,7 @@ def __init__(self,
 You can use native asynchronous version of the client:
 - https://influxdb-client.readthedocs.io/en/stable/usage.html#how-to-use-asyncio
         """
+# TODO above message has link to Influxdb2 API __NOT__ Influxdb3 API !!! - illustrates different API
             warnings.warn(message, DeprecationWarning)
 
     def write(self, bucket: str, org: str = None,

tests/test_api_client.py

@@ -1,8 +1,13 @@
 import json
+import time
 import unittest
 import uuid
+from concurrent.futures import thread
 from unittest import mock
+
+import pytest
 from urllib3 import response
+from urllib3.exceptions import ConnectTimeoutError
 
 from influxdb_client_3.write_client._sync.api_client import ApiClient
 from influxdb_client_3.write_client.configuration import Configuration
@@ -36,9 +41,20 @@ def data(self):
 
     return MockResponse(None, 200)
 
-
 class ApiClientTests(unittest.TestCase):
 
+    received_timeout_total = None
+    def mock_urllib3_timeout_request(method,
+                                     url,
+                                     body,
+                                     headers,
+                                     **urlopen_kw):
+        if urlopen_kw.get('timeout', None) is not None:
+            ApiClientTests.received_timeout_total = urlopen_kw['timeout'].total
+            raise ConnectTimeoutError()
+
+        return response.HTTPResponse(status=200, version=4, reason="OK", decode_content=False, request_url=url)
+
     def test_default_headers(self):
         global _package
         conf = Configuration()
@@ -140,10 +156,33 @@ def test_api_error_headers(self):
         self.assertEqual(headers['X-Influxdb-Request-Id'], requestid)
         self.assertEqual(headers['X-Influxdb-Build'], 'Mock')
 
-    def test_request_timeout_from_config(self):
-        # TODO
+    @mock.patch("urllib3._request_methods.RequestMethods.request",
+                side_effect=mock_urllib3_timeout_request)
+    def test_request_config_timeout(self, mock_request):
+        conf = Configuration()
+        conf.host = "http://localhost:8181"
+        conf.timeout = 300
+        local_client = ApiClient(conf)
+        service = WriteService(local_client)
+        with pytest.raises(ConnectTimeoutError):
+            service.post_write("TEST_ORG", "TEST_BUCKET", "data,foo=bar val=3.14",
+                              _preload_content=False)
+        self.assertEqual(0.3, self.received_timeout_total)
+        self.received_timeout_total = None
+
+    @mock.patch("urllib3._request_methods.RequestMethods.request",
+                side_effect=mock_urllib3_timeout_request)
+    def test_request_arg_timeout(self, mock_request):
         conf = Configuration()
+        conf.host = "http://localhost:8181"
+        conf.timeout = 300
         local_client = ApiClient(conf)
+        service = WriteService(local_client)
+        with pytest.raises(ConnectTimeoutError):
+            service.post_write("TEST_ORG", "TEST_BUCKET", "data,foo=bar val=3.14",
+                               _request_timeout=100, _preload_content=False)
+        self.assertEqual(0.1, self.received_timeout_total)
+        self.received_timeout_total = None
 
     def test_should_gzip(self):
         # Test when gzip is disabled

tests/test_influxdb_client_3_integration.py

@@ -7,7 +7,7 @@
 import time
 import unittest
 
-from urllib3.exceptions import MaxRetryError, ConnectTimeoutError
+from urllib3.exceptions import MaxRetryError, ConnectTimeoutError, TimeoutError as Url3TimeoutError
 
 from influxdb_client_3 import InfluxDBClient3, write_client_options, WriteOptions, SYNCHRONOUS, flight_client_options, \
     WriteType
@@ -251,22 +251,14 @@ def test_get_server_version(self):
         version = self.client.get_server_version()
         assert version is not None
 
-    # TODO set sync test, also investigate behavior with batcher and retry
-    # TODO do these need to be run with integration - won't mock suffice?
     def test_write_timeout_sync(self):
 
-        ErrorRecord = None
-        def set_error_record(error):
-            nonlocal ErrorRecord
-            ErrorRecord = error
-
-        with pytest.raises(ConnectTimeoutError) as e:
+        with pytest.raises(Url3TimeoutError):
             localClient = InfluxDBClient3(
                 host=self.host,
                 database=self.database,
                 token=self.token,
-                write_client_options=flight_client_options(
-                    error_callback=set_error_record,
+                write_client_options=write_client_options(
                     write_options=WriteOptions(
                         max_retry_time=0,
                         timeout=20,
@@ -277,83 +269,105 @@ def set_error_record(error):
 
             localClient.write("test_write_timeout,location=harfa fVal=3.14,iVal=42i")
 
-
-    @pytest.mark.skip(reason="placeholder - partially implemented")
     @asyncio_run
     async def test_write_timeout_async(self):
-        # fco = flight_client_options(max_retries=10, timeout=30_000)
-        # print(f"DEBUG fco: {fco}")
-        # TODO ensure API can handle either callback or thrown exception
-        # TODO asserts based on solution
 
-        ErrorRecord = None
-        def set_error_record(error):
-            nonlocal ErrorRecord
-            ErrorRecord = error
+        with pytest.raises(Url3TimeoutError):
+            localClient = InfluxDBClient3(
+                host=self.host,
+                database=self.database,
+                token=self.token,
+                write_client_options=write_client_options(
+                    # error_callback=set_error_record,
+                    write_options=WriteOptions(
+                        max_retry_time=0, # disable retries
+                        timeout=20,
+                        write_type=WriteType.asynchronous
+                    )
+                )
+            )
+
+            applyResult = localClient.write("test_write_timeout,location=harfa fVal=3.14,iVal=42i")
+            applyResult.get()
 
+    def test_write_timeout_batching(self):
+
+        ErrorResult = {"rt": None, "rd": None, "rx": None}
+
+        def set_error_result(rt, rd, rx):
+            nonlocal ErrorResult
+            ErrorResult = {"rt": rt, "rd": rd, "rx": rx}
 
         localClient = InfluxDBClient3(
             host=self.host,
             database=self.database,
             token=self.token,
-            write_client_options=flight_client_options(
-                error_callback=set_error_record,
+            write_client_options=write_client_options(
+                error_callback=set_error_result,
                 write_options=WriteOptions(
-                    max_retry_time=0,
+                    max_retry_time=0, # disable retries
                     timeout=20,
-                    write_type=WriteType.asynchronous
+                    write_type=WriteType.batching,
+                    max_retries=1,
+                    batch_size=1,
                 )
             )
         )
+        lp = "test_write_timeout,location=harfa fVal=3.14,iVal=42i"
+        localClient.write(lp)
 
-        print(f"DEBUG localClient._write_client_options: {localClient._write_client_options['write_options'].__dict__}")
-        print(f"DEBUG localClient._client._base._Configuration {localClient._client.conf.timeout}")
-
-        applyResult = localClient.write("test_write_timeout,location=harfa fVal=3.14,iVal=42i")
-        print(f"DEBUG applyResult: {applyResult}")
-        result = applyResult.get()
-        print(f"DEBUG result: {result}")
+        # wait for batcher attempt last write retry
+        time.sleep(0.1)
 
+        assert ErrorResult["rt"] == (self.database, 'default', 'ns')
+        assert ErrorResult["rd"] is not None
+        assert isinstance(ErrorResult["rd"], bytes)
+        assert ErrorResult["rd"].decode('utf-8') == lp
+        assert ErrorResult["rx"] is not None
+        assert isinstance(ErrorResult["rx"], MaxRetryError)
+        mre = ErrorResult["rx"]
+        assert isinstance(mre.reason, Url3TimeoutError)
 
-    def test_write_timeout_batching(self):
+    def test_write_timeout_retry(self):
 
         ErrorResult = {"rt": None, "rd": None, "rx": None}
-
         def set_error_result(rt, rd, rx):
             nonlocal ErrorResult
             ErrorResult = {"rt": rt, "rd": rd, "rx": rx}
 
+        retry_ct = 0
+        def retry_cb(args, data, excp):
+            nonlocal retry_ct
+            retry_ct += 1
+
         localClient = InfluxDBClient3(
             host=self.host,
             database=self.database,
             token=self.token,
-            write_client_options=flight_client_options(
+            write_client_options=write_client_options(
                 error_callback=set_error_result,
+                retry_callback=retry_cb,
                 write_options=WriteOptions(
-                    max_retry_time=0,
+                    max_retry_time=10000,
+                    max_retry_delay=100,
+                    retry_interval=100,
                     timeout=20,
-                    write_type=WriteType.batching,
-                    max_retries=1,
+                    max_retries=3,
                     batch_size=1,
                 )
             )
         )
+
         lp = "test_write_timeout,location=harfa fVal=3.14,iVal=42i"
         localClient.write(lp)
+        time.sleep(1) # await all retries
 
-        # wait for batcher attempt last write retry
-        time.sleep(0.1)
-
+        assert retry_ct == 3
         assert ErrorResult["rt"] == (self.database, 'default', 'ns')
         assert ErrorResult["rd"] is not None
         assert isinstance(ErrorResult["rd"], bytes)
         assert ErrorResult["rd"].decode('utf-8') == lp
         assert ErrorResult["rx"] is not None
         assert isinstance(ErrorResult["rx"], MaxRetryError)
         mre = ErrorResult["rx"]
-        assert isinstance(mre.reason, ConnectTimeoutError)
-
-    @pytest.mark.skip("place holder")
-    def test_write_timeout_retry(self):
-        # TODO
-        print("DEBUG test_write_timeout_retry")
+        assert isinstance(mre.reason, Url3TimeoutError)

tests/test_write_local_server.py

@@ -1,8 +1,10 @@
 import re
+import time
 from http import HTTPStatus
 
 import pytest
 from pytest_httpserver import HTTPServer, RequestMatcher
+from urllib3.exceptions import TimeoutError as urllib3_TimeoutError
 
 from influxdb_client_3 import InfluxDBClient3, WriteOptions, WritePrecision, write_client_options, WriteType
 from influxdb_client_3.write_client.rest import ApiException
@@ -20,6 +22,10 @@ def assert_request_made(httpserver, matcher):
         httpserver.assert_request_made(matcher)
         httpserver.check_assertions()
 
+    @staticmethod
+    def delay_response(httpserver, delay = 1.0):
+        httpserver.expect_request(re.compile(".*")).respond_with_handler(lambda request: time.sleep(delay))
+
     def test_write_default_params(self, httpserver: HTTPServer):
         self.set_response_status(httpserver, 200)
 
@@ -127,3 +133,20 @@ def test_write_with_no_sync_true_and_gzip(self, httpserver: HTTPServer):
             method="POST", uri="/api/v3/write_lp",
             query_string={"org": "ORG", "db": "DB", "precision": "microsecond", "no_sync": "true"},
             headers={"Content-Encoding": "gzip"}, ))
+
+    def test_write_with_timeout(self, httpserver: HTTPServer):
+        self.delay_response(httpserver, 0.5)
+
+        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,
+                        timeout=30,
+                        no_sync=True
+                    )
+                ),
+                enable_gzip=True
+            ).write(self.SAMPLE_RECORD)