Merge branch 'main' into typed-definition-setup.py

CHANGELOG.md

@@ -2,8 +2,14 @@
 
 ## 0.9.0 [unreleased]
 
+### Features
+
+1. [#108](https://github.com/InfluxCommunity/influxdb3-python/pull/108): Better expose access to response headers in `InfluxDBError`.  Example `handle_http_error` added.
+
 ### Bug Fixes
+
 1. [#107](https://github.com/InfluxCommunity/influxdb3-python/pull/107): Missing `py.typed` in distribution package
+1. [#111](https://github.com/InfluxCommunity/influxdb3-python/pull/111): Reduce log level of disposal of batch processor to DEBUG
 
 ## 0.8.0 [2024-08-12]
 

Examples/__init__.py

@@ -0,0 +1 @@
+# used mainly to resolve local utility helpers like config.py

Examples/cloud_dedicated_query.py

@@ -1,10 +1,13 @@
+from config import Config
 import influxdb_client_3 as InfluxDBClient3
 
+config = Config()
+
 client = InfluxDBClient3.InfluxDBClient3(
-    token="",
-    host="b0c7cce5-8dbc-428e-98c6-7f996fb96467.a.influxdb.io",
-    org="6a841c0c08328fb1",
-    database="flight2")
+    token=config.token,
+    host=config.host,
+    org=config.org,
+    database=config.database)
 
 table = client.query(
     query="SELECT * FROM flight WHERE time > now() - 4h",

Examples/cloud_dedicated_write.py

@@ -1,16 +1,17 @@
-
+from config import Config
 import influxdb_client_3 as InfluxDBClient3
-from influxdb_client_3 import write_options
+from influxdb_client_3 import WriteOptions
 import pandas as pd
 import numpy as np
 
+config = Config()
 
 client = InfluxDBClient3.InfluxDBClient3(
-    token="",
-    host="b0c7cce5-8dbc-428e-98c6-7f996fb96467.a.influxdb.io",
-    org="6a841c0c08328fb1",
-    database="flight2",
-    write_options=write_options(
+    token=config.token,
+    host=config.host,
+    org=config.org,
+    database=config.database,
+    write_options=WriteOptions(
         batch_size=500,
         flush_interval=10_000,
         jitter_interval=2_000,
@@ -19,15 +20,15 @@
         max_retry_delay=30_000,
         max_close_wait=300_000,
         exponential_base=2,
-         write_type='batching'))
+        write_type='batching'))
 
 
 # Create a dataframe
 df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
 
 
 # Create a range of datetime values
-dates = pd.date_range(start='2023-05-01', end='2023-05-29', freq='5min')
+dates = pd.date_range(start='2024-09-08', end='2024-09-09', freq='5min')
 
 # Create a DataFrame with random data and datetime index
 df = pd.DataFrame(

Examples/config.py

@@ -0,0 +1,13 @@
+import os
+import json
+
+
+class Config:
+    def __init__(self):
+        self.host = os.getenv('INFLUXDB_HOST') or 'https://us-east-1-1.aws.cloud2.influxdata.com/'
+        self.token = os.getenv('INFLUXDB_TOKEN') or 'my-token'
+        self.org = os.getenv('INFLUXDB_ORG') or 'my-org'
+        self.database = os.getenv('INFLUXDB_DATABASE') or 'my-db'
+
+    def __str__(self):
+        return json.dumps(self.__dict__)

Examples/handle_http_error.py

@@ -0,0 +1,43 @@
+"""
+Demonstrates handling response error headers on error.
+"""
+import logging
+from config import Config
+
+import influxdb_client_3 as InfluxDBClient3
+
+
+def main() -> None:
+    """
+    Main function
+    :return:
+    """
+    config = Config()
+    logging.basicConfig(format='%(asctime)s %(message)s', level=logging.INFO)
+
+    client = InfluxDBClient3.InfluxDBClient3(
+        host=config.host,
+        token=config.token,
+        org=config.org,
+        database=config.database
+    )
+
+    # write with empty field results in HTTP 400 error
+    # Other cases might be HTTP 503 or HTTP 429 too many requests
+    lp = 'drone,location=harfa,id=A16E22 speed=18.7,alt=97.6,shutter='
+
+    try:
+        client.write(lp)
+    except InfluxDBClient3.InfluxDBError as idberr:
+        logging.log(logging.ERROR, 'WRITE ERROR: %s (%s)',
+                    idberr.response.status,
+                    idberr.message)
+        headers_string = 'Response Headers:\n'
+        headers = idberr.getheaders()
+        for h in headers:
+            headers_string += f'   {h}: {headers[h]}\n'
+        logging.log(logging.INFO, headers_string)
+
+
+if __name__ == "__main__":
+    main()

influxdb_client_3/write_client/client/exceptions.py

@@ -50,3 +50,7 @@ def get(d, key):
 
         # Http Status
         return response.reason
+
+    def getheaders(self):
+        """Helper method to make response headers more accessible."""
+        return self.response.getheaders()

influxdb_client_3/write_client/client/write_api.py

@@ -566,7 +566,7 @@ def _on_error(ex):
 
     def _on_complete(self):
         self._disposable.dispose()
-        logger.info("the batching processor was disposed")
+        logger.debug("the batching processor was disposed")
 
     def __getstate__(self):
         """Return a dict of attributes that you want to pickle."""

tests/test_api_client.py

@@ -1,4 +1,6 @@
+import json
 import unittest
+import uuid
 from unittest import mock
 from urllib3 import response
 
@@ -105,3 +107,35 @@ def test_api_error_unknown(self):
         with self.assertRaises(InfluxDBError) as err:
             self._test_api_error(response_body)
         self.assertEqual(response_body, err.exception.message)
+
+    def test_api_error_headers(self):
+        body = '{"error": "test error"}'
+        body_dic = json.loads(body)
+        conf = Configuration()
+        local_client = ApiClient(conf)
+        traceid = "123456789ABCDEF0"
+        requestid = uuid.uuid4().__str__()
+
+        local_client.rest_client.pool_manager.request = mock.Mock(
+            return_value=response.HTTPResponse(
+                status=400,
+                reason='Bad Request',
+                headers={
+                    'Trace-Id': traceid,
+                    'Trace-Sampled': 'false',
+                    'X-Influxdb-Request-Id': requestid,
+                    'X-Influxdb-Build': 'Mock'
+                },
+                body=body.encode()
+            )
+        )
+        with self.assertRaises(InfluxDBError) as err:
+            service = WriteService(local_client)
+            service.post_write("TEST_ORG", "TEST_BUCKET", "data,foo=bar val=3.14")
+        self.assertEqual(body_dic['error'], err.exception.message)
+        headers = err.exception.getheaders()
+        self.assertEqual(4, len(headers))
+        self.assertEqual(headers['Trace-Id'], traceid)
+        self.assertEqual(headers['Trace-Sampled'], 'false')
+        self.assertEqual(headers['X-Influxdb-Request-Id'], requestid)
+        self.assertEqual(headers['X-Influxdb-Build'], 'Mock')

tests/test_influxdb_client_3_integration.py

@@ -56,3 +56,19 @@ def test_auth_error_auth_scheme(self):
         with self.assertRaises(InfluxDBError) as err:
             self.client.write(f"integration_test_python,type=used value=123.0,test_id={test_id}i")
         self.assertEqual('unauthorized access', err.exception.message)  # Cloud
+
+    def test_error_headers(self):
+        self.client = InfluxDBClient3(host=self.host, database=self.database, token=self.token)
+        with self.assertRaises(InfluxDBError) as err:
+            self.client.write("integration_test_python,type=used value=123.0,test_id=")
+        self.assertIn("Could not parse entire line. Found trailing content:", err.exception.message)
+        headers = err.exception.getheaders()
+        try:
+            self.assertIsNotNone(headers)
+            self.assertRegex(headers['trace-id'], '[0-9a-f]{16}')
+            self.assertEqual('false', headers['trace-sampled'])
+            self.assertIsNotNone(headers['Strict-Transport-Security'])
+            self.assertRegex(headers['X-Influxdb-Request-ID'], '[0-9a-f]+')
+            self.assertIsNotNone(headers['X-Influxdb-Build'])
+        except KeyError as ke:
+            self.fail(f'Header {ke} not found')