feat: (WIP) adds example error_handling.py and tests of headers in error types

Author: karel-rehor

examples/error_handling.py

@@ -0,0 +1,86 @@
+import asyncio
+import time
+from datetime import datetime, timezone
+
+from influxdb_client.client.exceptions import InfluxDBError
+
+from influxdb_client.rest import ApiException
+
+from influxdb_client.client.write_api import SYNCHRONOUS
+
+from influxdb_client import WritePrecision, InfluxDBClient, Point
+from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
+
+
+class BatchCB(object):
+
+    @staticmethod
+    def success(self, conf: (str, str, str), data: str):
+        print(f"Write success: {conf}, data: {data}")
+
+    @staticmethod
+    def error(self, conf: (str, str, str), data: str, exception: InfluxDBError):
+        print(f"Write failed: {conf}, data: {data}, error: {exception.message}")
+        print(f"   Date:                  {exception.headers.get("Date")}")
+        print(f"   X-Influxdb-Build:      {exception.headers.get("X-Influxdb-Build")}")
+        print(f"   X-Influxdb-Version:    {exception.headers.get("X-Influxdb-Version")}")
+        print(f"   X-Platform-Error-Code: {exception.headers.get("X-Platform-Error-Code")}")
+        print(f"   Retry-After:           {exception.headers.get("Retry-After")}")
+
+    @staticmethod
+    def retry(self, conf: (str, str, str), data: str, exception: InfluxDBError):
+        print(f"Write failed but retryable: {conf}, data: {data}, error: {exception}")
+
+
+def report_ping(ping: bool):
+    if not ping:
+        raise ValueError("InfluxDB: Failed to ping server")
+    else:
+        print("InfluxDB: ready")
+
+
+def use_sync():
+    print("Using sync")
+    with InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") as client:
+        report_ping(client.ping())
+        try:
+            client.write_api(write_options=SYNCHRONOUS).write(bucket="my-bucket", record="cpu,location=G4 usage=")
+        except ApiException as ae:
+            print("\nCaught ae: ", ae.message)
+            print("   Date:                  ", ae.headers.get("Date"))
+            print("   X-Influxdb-Build:      ", ae.headers.get("X-Influxdb-Build"))
+            print("   X-Influxdb-Version:    ", ae.headers.get("X-Influxdb-Version"))
+            print("   X-Platform-Error-Code: ", ae.headers.get("X-Platform-Error-Code"))
+            print("   Retry-After:           ", ae.headers.get("Retry-After"))  # Should be None
+
+        print("Sync write done")
+
+
+def use_batch():
+    print("Using batch")
+    with InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") as client:
+        cb = BatchCB()
+        with client.write_api(success_callback=cb.success,
+                              error_callback=cb.error,
+                              retry_callback=cb.retry) as write_api:
+            write_api.write(bucket="my-bucket", record="cpu,location=G9 usage=")
+            print("Batch write sent")
+
+
+async def use_async():
+    print("Using async")
+    async with InfluxDBClientAsync(url="http://localhost:8086", token="my-token", org="my-org") as client:
+        report_ping(await client.ping())
+        try:
+            await client.write_api().write(bucket="my-bucket", record="cpu,location=G7 usage=")
+        except InfluxDBError as ie:
+            print("\nCaught ie: ", ie)
+        print("Async write done")
+
+
+if __name__ == "__main__":
+    use_sync()
+    print("\n   Continuing...")
+    use_batch()
+    print("\n   Continuing...")
+    asyncio.run(use_async())

influxdb_client/client/exceptions.py

@@ -16,8 +16,10 @@ def __init__(self, response: HTTPResponse = None, message: str = None):
             self.response = response
             self.message = self._get_message(response)
             if isinstance(response, HTTPResponse):  # response is HTTPResponse
+                self.headers = response.headers
                 self.retry_after = response.headers.get('Retry-After')
             else:  # response is RESTResponse
+                self.headers = response.getheaders()
                 self.retry_after = response.getheader('Retry-After')
         else:
             self.response = None

tests/test_InfluxDBClientAsync.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import re
 import unittest
 import os
 from datetime import datetime, timezone
@@ -389,6 +390,25 @@ async def test_query_exception_propagation(self):
         with pytest.raises(InfluxDBError) as e:
             await self.client.query_api().query("buckets()", "my-org")
         self.assertEqual("unauthorized access", e.value.message)
+        print("DEBUG e.headers: ", e.value.headers)
+
+    @async_test
+    async def test_write_exception_propagation(self):
+        await self.client.close()
+        self.client = InfluxDBClientAsync(url="http://localhost:8086", token="wrong", org="my-org")
+
+        with pytest.raises(InfluxDBError) as e:
+            await self.client.write_api().write(bucket="my_bucket",
+                                                record="temperature,location=hic cels=")
+        self.assertEqual("unauthorized access", e.value.message)
+        headers = e.value.headers
+        self.assertIsNotNone(headers)
+        self.assertIsNotNone(headers.get("Content-Length"))
+        self.assertIsNotNone(headers.get("Date"))
+        self.assertIsNotNone(headers.get("X-Platform-Error-Code"))
+        self.assertIn("application/json", headers.get("Content-Type"))
+        self.assertTrue(re.compile("^v.*").match(headers.get("X-Influxdb-Version")))
+        self.assertEqual("OSS", headers.get("X-Influxdb-Build"))
 
     @async_test
     @aioresponses()

tests/test_WriteApi.py

@@ -3,12 +3,16 @@
 from __future__ import absolute_import
 
 import datetime
+import json
+import logging
 import os
+import re
 import sys
 import unittest
 from collections import namedtuple
 from datetime import timedelta
 from multiprocessing.pool import ApplyResult
+from types import SimpleNamespace
 
 import httpretty
 import pytest
@@ -190,6 +194,17 @@ def test_write_error(self):
 
         self.assertEqual(400, exception.status)
         self.assertEqual("Bad Request", exception.reason)
+        # assert headers
+        self.assertIsNotNone(exception.headers)
+        self.assertIsNotNone(exception.headers.get("Content-Length"))
+        self.assertIsNotNone(exception.headers.get("Date"))
+        self.assertIsNotNone(exception.headers.get("X-Platform-Error-Code"))
+        self.assertIn("application/json", exception.headers.get("Content-Type"))
+        self.assertTrue(re.compile("^v.*").match(exception.headers.get("X-Influxdb-Version")))
+        self.assertEqual("OSS", exception.headers.get("X-Influxdb-Build"))
+        # assert body
+        b = json.loads(exception.body, object_hook=lambda d: SimpleNamespace(**d))
+        self.assertTrue(re.compile("^unable to parse.*invalid field format").match(b.message))
 
     def test_write_dictionary(self):
         _bucket = self.create_test_bucket()
@@ -609,6 +624,28 @@ def test_write_result(self):
         self.assertEqual(None, result.get())
         self.delete_test_bucket(_bucket)
 
+    def test_write_error(self):
+        _bucket = self.create_test_bucket()
+
+        _record = "h2o_feet,location=coyote_creek level\\ water_level="
+        result = self.write_client.write(_bucket.name, self.org, _record)
+
+        with self.assertRaises(ApiException) as cm:
+            result.get()
+        self.assertEqual(400, cm.exception.status)
+        self.assertEqual("Bad Request", cm.exception.reason)
+        # assert headers
+        self.assertIsNotNone(cm.exception.headers)
+        self.assertIsNotNone(cm.exception.headers.get("Content-Length"))
+        self.assertIsNotNone(cm.exception.headers.get("Date"))
+        self.assertIsNotNone(cm.exception.headers.get("X-Platform-Error-Code"))
+        self.assertIn("application/json", cm.exception.headers.get("Content-Type"))
+        self.assertTrue(re.compile("^v.*").match(cm.exception.headers.get("X-Influxdb-Version")))
+        self.assertEqual("OSS", cm.exception.headers.get("X-Influxdb-Build"))
+        # assert body
+        b = json.loads(cm.exception.body, object_hook=lambda d: SimpleNamespace(**d))
+        self.assertTrue(re.compile("^unable to parse.*missing field value").match(b.message))
+
     def test_write_dictionaries(self):
         bucket = self.create_test_bucket()