chore: add support to configure retries

Author: bednar

.circleci/config.yml

@@ -130,6 +130,7 @@ jobs:
             pip install -e . --user
             pip install -e .\[async\] --user
             pip install -e .\[extra\] --user
+            pip install aiohttp-retry --user
             export PYTHONPATH="$PWD"
             cd examples
             python ./monitoring_and_alerting.py
@@ -139,6 +140,7 @@ jobs:
             python ./asynchronous.py
             python ./asynchronous_management.py
             python ./asynchronous_batching.py
+            python ./asynchronous_retry.py
   check-sphinx:
     docker:
       - image: *default-python

examples/README.md

@@ -33,5 +33,6 @@
 ## Asynchronous
 - [asynchronous.py](asynchronous.py) - How to use Asyncio with InfluxDB client
 - [asynchronous_management.py](asynchronous_management.py) - How to use asynchronous Management API
-- [asynchronous_batching.py](asynchronous_batching.py) - HHow to use [RxPY](https://rxpy.readthedocs.io/en/latest/) to prepare batches for Asyncio client
+- [asynchronous_batching.py](asynchronous_batching.py) - How to use [RxPY](https://rxpy.readthedocs.io/en/latest/) to prepare batches
+- [asynchronous_retry.py](asynchronous_retry.py) - How to use [aiohttp-retry](https://github.com/inyutin/aiohttp_retry) to configure retries
 

examples/asynchronous_retry.py

@@ -0,0 +1,47 @@
+"""
+How to use `aiohttp-retry` with async client.
+
+This example depends on `aiohttp_retry <https://github.com/inyutin/aiohttp_retry>`_.
+Install ``aiohttp_retry`` by: pip install aiohttp-retry.
+
+"""
+import asyncio
+
+from aiohttp_retry import ExponentialRetry, RetryClient
+
+from influxdb_client import Point
+from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
+
+
+async def main():
+    """
+    Configure Retries - for more info see https://github.com/inyutin/aiohttp_retry
+    """
+    retry_options = ExponentialRetry(attempts=3)
+    async with InfluxDBClientAsync(url="http://localhost:8086", token="my-token", org="my-org",
+                                   client_session_type=RetryClient,
+                                   client_session_kwargs={"retry_options": retry_options}) as client:
+        """
+        Write data:
+        """
+        print(f"\n------- Written data: -------\n")
+        write_api = client.write_api()
+        _point1 = Point("async_m").tag("location", "Prague").field("temperature", 25.3)
+        _point2 = Point("async_m").tag("location", "New York").field("temperature", 24.3)
+        successfully = await write_api.write(bucket="my-bucket", record=[_point1, _point2])
+        print(f" > successfully: {successfully}")
+
+        """
+        Query: Stream of FluxRecords
+        """
+        print(f"\n------- Query: Stream of FluxRecords -------\n")
+        query_api = client.query_api()
+        records = await query_api.query_stream('from(bucket:"my-bucket") '
+                                               '|> range(start: -10m) '
+                                               '|> filter(fn: (r) => r["_measurement"] == "async_m")')
+        async for record in records:
+            print(record)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

influxdb_client/_async/rest.py

@@ -122,11 +122,14 @@ def __init__(self, configuration, pools_size=4, maxsize=None, **kwargs):
             timeout = aiohttp.client.DEFAULT_TIMEOUT
 
         # https pool manager
-        self.pool_manager = aiohttp.ClientSession(
+        _client_session_type = kwargs.get('client_session_type', aiohttp.ClientSession)
+        _client_session_kwargs = kwargs.get('client_session_kwargs', {})
+        self.pool_manager = _client_session_type(
             connector=connector,
             trust_env=True,
             timeout=timeout,
-            trace_configs=[trace_config] if configuration.debug else None
+            trace_configs=[trace_config] if configuration.debug else None,
+            **_client_session_kwargs
         )
 
     async def close(self):

influxdb_client/client/influxdb_client_async.py

@@ -40,6 +40,9 @@ def __init__(self, url, token, org: str = None, debug=None, timeout=10_000, enab
                               (defaults to false, don't set to true when talking to InfluxDB 2)
         :key bool allow_redirects: If set to ``False``, do not follow HTTP redirects. ``True`` by default.
         :key int max_redirects: Maximum number of HTTP redirects to follow. ``10`` by default.
+        :key dict client_session_kwargs: Additional configuration arguments for :class:`~aiohttp.ClientSession`
+        :key type client_session_type: Type of aiohttp client to use. Useful for third party wrappers like
+                                       ``aiohttp-retry``. :class:`~aiohttp.ClientSession` by default.
         :key list[str] profilers: list of enabled Flux profilers
         """
         super().__init__(url=url, token=token, org=org, debug=debug, timeout=timeout, enable_gzip=enable_gzip, **kwargs)