feat: (WIP) - placeholder. Adds timeout property to WriteOptions and proposed tests

Author: karel-rehor

influxdb_client_3/__init__.py

@@ -220,11 +220,13 @@ def __init__(
  write_type = DefaultWriteOptions.write_type.value
  write_precision = DefaultWriteOptions.write_precision.value
  write_no_sync = DefaultWriteOptions.no_sync.value
+ write_timeout = DefaultWriteOptions.timeout.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_timeout = getattr(write_opts, 'timeout', write_timeout)
 
  write_options = WriteOptions(
  write_type=write_type,
@@ -253,6 +255,7 @@ def __init__(
  url=f"{scheme}://{hostname}:{port}",
  token=self._token,
  org=self._org,
+ timeout=write_timeout,
  **kwargs)
 
  self._write_api = _WriteApi(influxdb_client=self._client, **self._write_client_options)
@@ -344,6 +347,7 @@ def write(self, record=None, database=None, **kwargs):
  :type database: str
  :param kwargs: Additional arguments to pass to the write API.
  """
+ print(f"DEBUG InfluxDBClient3.write {record}")
  if database is None:
  database = self._database
 

influxdb_client_3/write_client/_sync/api_client.py

@@ -357,14 +357,18 @@ def call_api(self, resource_path, method,
  If parameter async_req is False or missing,
  then the method will return the response directly.
  """
+ print("DEBUG ApiClient.call_api()")
  if not async_req:
+ print(" DEBUG synchronous call")
  return self.__call_api(resource_path, method,
  path_params, query_params, header_params,
  body, post_params, files,
  response_type, auth_settings,
  _return_http_data_only, collection_formats,
  _preload_content, _request_timeout, urlopen_kw)
  else:
+ # TODO possible refactor - async handler inside package `_sync`?
+ print(" DEBUG asynchronous call")
  thread = self.pool.apply_async(self.__call_api, (resource_path,
  method, path_params, query_params,
  header_params, body,

influxdb_client_3/write_client/client/write_api.py

@@ -25,6 +25,7 @@
 from influxdb_client_3.write_client.rest import _UTF_8_encoding
 
 DEFAULT_WRITE_NO_SYNC = False
+DEFAULT_WRITE_TIMEOUT = 10_000
 
 logger = logging.getLogger('influxdb_client_3.write_client.client.write_api')
 
@@ -45,6 +46,7 @@ class DefaultWriteOptions(Enum):
  write_type = WriteType.synchronous
  write_precision = DEFAULT_WRITE_PRECISION
  no_sync = DEFAULT_WRITE_NO_SYNC
+ timeout = DEFAULT_WRITE_TIMEOUT
 
 
 class WriteOptions(object):
@@ -61,6 +63,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
  max_close_wait=300_000,
  write_precision=DEFAULT_WRITE_PRECISION,
  no_sync=DEFAULT_WRITE_NO_SYNC,
+ timeout=DEFAULT_WRITE_TIMEOUT,
  write_scheduler=ThreadPoolScheduler(max_workers=1)) -> None:
  """
  Create write api configuration.
@@ -93,6 +96,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.timeout = timeout
  self.no_sync = no_sync
 
  def to_retry_strategy(self, **kwargs):
@@ -268,6 +272,8 @@ def __init__(self,
  # Define Subject that listen incoming data and produces writes into InfluxDB
  self._subject = Subject()
 
+ print(f"DEBUG batching write with subject {self._subject}")
+
  self._disposable = self._subject.pipe(
  # Split incoming data to windows by batch_size or flush_interval
  ops.window_with_time_or_count(count=write_options.batch_size,
@@ -372,6 +378,7 @@ def write(self, bucket: str, org: str = None,
 
  """ # noqa: E501
  org = get_org_query_param(org=org, client=self._influxdb_client)
+ print("DEBUG WriteApi.write()")
 
  self._append_default_tags(record)
 
@@ -390,13 +397,21 @@ def write(self, bucket: str, org: str = None,
  _async_req = True if self._write_options.write_type == WriteType.asynchronous else False
 
  def write_payload(payload):
+ print("DEBUG WriteApi.write_payload()")
  final_string = b'\n'.join(payload[1])
- return self._post_write(_async_req, bucket, org, final_string, payload[0], no_sync)
+ result = self._post_write(_async_req, bucket, org, final_string, payload[0], no_sync)
+ # return self._post_write(_async_req, bucket, org, final_string, payload[0], no_sync)
+ print(f" DEBUG write_payload() result {result}")
+ return result
 
  results = list(map(write_payload, payloads.items()))
+ print(f" DEBUG WriteApi.write() results {results}")
  if not _async_req:
+ print(f" --->DEBUG not async_request")
  return None
  elif len(results) == 1:
+ # TODO if results contains error or exception ensure handled
+ print(f" --->DEBUG async_request() results {results[0]}")
  return results[0]
  return results
 
@@ -464,11 +479,13 @@ def __del__(self):
  def _write_batching(self, bucket, org, data,
  precision=None,
  **kwargs):
+ print("DEBUG _write_batching()")
  if precision is None:
  precision = self._write_options.write_precision
 
  if isinstance(data, bytes):
  _key = _BatchItemKey(bucket, org, precision)
+ print(f" DEBUG _write_batching() data bytes {_key}")
  self._subject.on_next(_BatchItem(key=_key, data=data))
 
  elif isinstance(data, str):
@@ -520,6 +537,8 @@ def _write_batching(self, bucket, org, data,
 
  def _http(self, batch_item: _BatchItem):
 
+ print("DEBUG _http()")
+
  logger.debug("Write time series data into InfluxDB: %s", batch_item)
 
  if self._retry_callback:
@@ -540,6 +559,7 @@ def _retry_callback_delegate(exception):
  return _BatchResponse(data=batch_item)
 
  def _post_write(self, _async_req, bucket, org, body, precision, no_sync, **kwargs):
+ print("DEBUG write_api._post_write()")
 
  return self._write_service.post_write(org=org, bucket=bucket, body=body, precision=precision,
  no_sync=no_sync,
@@ -549,6 +569,8 @@ def _post_write(self, _async_req, bucket, org, body, precision, no_sync, **kwarg
 
  def _to_response(self, data: _BatchItem, delay: timedelta):
 
+ print("DEBUG _to_response()")
+
  return rx.of(data).pipe(
  ops.subscribe_on(self._write_options.write_scheduler),
  # use delay if its specified

influxdb_client_3/write_client/service/write_service.py

@@ -42,9 +42,15 @@ def post_write(self, org, bucket, body, **kwargs): # noqa: E501,D401,D403
  returns the request thread.
  """ # noqa: E501
  kwargs['_return_http_data_only'] = True
+ print("DEBUG WriteService.post_write()")
  if kwargs.get('async_req'):
- return self.post_write_with_http_info(org, bucket, body, **kwargs) # noqa: E501
+ print(" DEBUG making asynchronous request ")
+ thread = self.post_write_with_http_info(org, bucket, body, **kwargs) # noqa: E501
+ print(f" DEBUG thread: {thread} ")
+ # return self.post_write_with_http_info(org, bucket, body, **kwargs) # noqa: E501
+ return thread
  else:
+ print(" DEBUG making synchronous request ")
  (data) = self.post_write_with_http_info(org, bucket, body, **kwargs) # noqa: E501
  return data
 
@@ -72,10 +78,14 @@ def post_write_with_http_info(self, org, bucket, body, **kwargs): # noqa: E501,
  :return: None
  If the method is called asynchronously,
  returns the request thread.
- """ # noqa: E501
+ """
+ print("WriteService.post_write_with_http_info()")
+ # noqa: E501
  local_var_params, path, path_params, query_params, header_params, body_params = \
  self._post_write_prepare(org, bucket, body, **kwargs) # noqa: E501
 
+ print(f"DEBUG local_var_params: {local_var_params}")
+
  try:
  return self.api_client.call_api(
  path, 'POST',
@@ -146,6 +156,8 @@ async def post_write_async(self, org, bucket, body, **kwargs): # noqa: E501,D40
  def _post_write_prepare(self, org, bucket, body, **kwargs): # noqa: E501,D401,D403
  local_var_params = dict(locals())
 
+ print(f"DEBUG local_var_params: {local_var_params}")
+
  all_params = ['org', 'bucket', 'body', 'zap_trace_span', 'content_encoding', 'content_type', 'content_length', 'accept', 'org_id', 'precision', 'no_sync'] # noqa: E501
  self._check_operation_params('post_write', all_params, local_var_params)
  # verify the required parameter 'org' is set

tests/test_api_client.py

@@ -140,6 +140,11 @@ 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
+ conf = Configuration()
+ local_client = ApiClient(conf)
+
  def test_should_gzip(self):
  # Test when gzip is disabled
  self.assertFalse(ApiClient.should_gzip("test", enable_gzip=False, gzip_threshold=1))

tests/test_influxdb_client_3.py

@@ -4,7 +4,8 @@
 
 from pytest_httpserver import HTTPServer
 
-from influxdb_client_3 import InfluxDBClient3, WritePrecision, DefaultWriteOptions, Point, WriteOptions, WriteType
+from influxdb_client_3 import InfluxDBClient3, WritePrecision, DefaultWriteOptions, Point, WriteOptions, WriteType, \
+ write_client_options
 from influxdb_client_3.exceptions import InfluxDB3ClientQueryError
 from influxdb_client_3.write_client.rest import ApiException
 from tests.util import asyncio_run
@@ -66,6 +67,51 @@ def test_token_auth_scheme_explicit(self):
  )
  self.assertEqual(client._client.auth_header_value, "my_scheme my_token")
 
+ def test_write_options(self):
+ client = InfluxDBClient3(
+ host="localhost",
+ org="my_org",
+ token="my_token",
+ auth_scheme="my_scheme",
+ write_client_options=write_client_options(
+ success_callback=lambda _: True,
+ error_callback=lambda _: False,
+ extra_arg="ignored",
+ write_options=WriteOptions(write_type=WriteType.synchronous,
+ max_retries=0,
+ max_retry_time=0,
+ max_retry_delay=0,
+ timeout=30_000,
+ flush_interval=500,))
+ )
+
+ assert isinstance(client._write_client_options["write_options"], WriteOptions)
+ assert client._write_client_options["success_callback"]("an_arg") == True
+ assert client._write_client_options["error_callback"]("an_arg") == False
+ assert client._write_client_options["extra_arg"] == "ignored"
+ assert client._write_client_options["write_options"].timeout == 30_000
+ assert client._write_client_options["write_options"].max_retries == 0
+ assert client._write_client_options["write_options"].max_retry_time == 0
+ assert client._write_client_options["write_options"].max_retry_delay == 0
+ assert client._write_client_options["write_options"].write_type == WriteType.synchronous
+ assert client._write_client_options["write_options"].flush_interval == 500
+ print(f"DEBUG client._client._base._Configuration {client._client.conf.__dict__}")
+ print(f"DEBUG client._client._base._Configuration.timeout {client._client.conf.timeout}")
+
+
+ def test_default_write_options(self):
+ client = InfluxDBClient3(
+ host="localhost",
+ token="my_token",
+ org="my_org",
+ database="my_db",
+ )
+
+ assert client._write_client_options["write_options"].write_type == DefaultWriteOptions.write_type.value
+ assert client._write_client_options["write_options"].no_sync == DefaultWriteOptions.no_sync.value
+ assert client._write_client_options["write_options"].write_precision == DefaultWriteOptions.write_precision.value
+ assert client._write_client_options["write_options"].timeout == DefaultWriteOptions.timeout.value
+
  @asyncio_run
  async def test_query_async(self):
  with ConstantFlightServer() as server:

tests/test_influxdb_client_3_integration.py

@@ -7,7 +7,10 @@
 import time
 import unittest
 
-from influxdb_client_3 import InfluxDBClient3, write_client_options, WriteOptions
+from urllib3.exceptions import MaxRetryError, ConnectTimeoutError
+
+from influxdb_client_3 import InfluxDBClient3, write_client_options, WriteOptions, SYNCHRONOUS, flight_client_options, \
+ WriteType
 from influxdb_client_3.exceptions import InfluxDBError
 from tests.util import asyncio_run, lp_to_py_object
 
@@ -247,3 +250,110 @@ async def test_verify_query_async(self):
  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:
+ localClient = InfluxDBClient3(
+ host=self.host,
+ database=self.database,
+ token=self.token,
+ write_client_options=flight_client_options(
+ error_callback=set_error_record,
+ write_options=WriteOptions(
+ max_retry_time=0,
+ timeout=20,
+ write_type=WriteType.synchronous
+ )
+ )
+ )
+
+ 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
+
+
+ localClient = InfluxDBClient3(
+ host=self.host,
+ database=self.database,
+ token=self.token,
+ write_client_options=flight_client_options(
+ error_callback=set_error_record,
+ write_options=WriteOptions(
+ max_retry_time=0,
+ timeout=20,
+ write_type=WriteType.asynchronous
+ )
+ )
+ )
+
+ 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}")
+
+
+ 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_result,
+ write_options=WriteOptions(
+ max_retry_time=0,
+ timeout=20,
+ write_type=WriteType.batching,
+ max_retries=1,
+ batch_size=1,
+ )
+ )
+ )
+ lp = "test_write_timeout,location=harfa fVal=3.14,iVal=42i"
+ localClient.write(lp)
+
+ # 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, ConnectTimeoutError)
+
+ @pytest.mark.skip("place holder")
+ def test_write_timeout_retry(self):
+ # TODO
+ print("DEBUG test_write_timeout_retry")