Merge branch 'influxdata:master' into nan-error-field-starting-with-digit

Author: fdorssers

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 :warning: This release drop supports for Python 3.6. As of 2021-12-23, 3.6 has reached the end-of-life phase of its release cycle. 3.6.15 was the final security release. For more info see: https://peps.python.org/pep-0494/#lifespan
 
+### Bug Fixes
+1. [#483](https://github.com/influxdata/influxdb-client-python/pull/483): Querying data if the `debug` is enabled
+1. [#477](https://github.com/influxdata/influxdb-client-python/pull/477): Parsing date fails due to thread race
+
 ### Dependencies
 1. [#472](https://github.com/influxdata/influxdb-client-python/pull/472): Update `RxPY` to `4.0.4`
 

influxdb_client/client/flux_csv_parser.py

@@ -85,7 +85,12 @@ def _close(self):
 
     def __enter__(self):
         """Initialize CSV reader."""
-        self._reader = csv_parser.reader(codecs.iterdecode(self._response, _UTF_8_encoding))
+        # response can be exhausted by logger, so we have to use data that has already been read
+        if hasattr(self._response, 'closed') and self._response.closed:
+            from io import StringIO
+            self._reader = csv_parser.reader(StringIO(self._response.data.decode(_UTF_8_encoding)))
+        else:
+            self._reader = csv_parser.reader(codecs.iterdecode(self._response, _UTF_8_encoding))
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):

influxdb_client/client/util/date_utils.py

@@ -1,11 +1,14 @@
 """Utils to get right Date parsing function."""
 import datetime
+import threading
 from datetime import timezone as tz
 
 from dateutil import parser
 
 date_helper = None
 
+lock_ = threading.Lock()
+
 
 class DateHelper:
     """
@@ -79,11 +82,15 @@ def get_date_helper() -> DateHelper:
     """
     global date_helper
     if date_helper is None:
-        date_helper = DateHelper()
-        try:
-            import ciso8601
-            date_helper.parse_date = ciso8601.parse_datetime
-        except ModuleNotFoundError:
-            date_helper.parse_date = parser.parse
+        with lock_:
+            # avoid duplicate initialization
+            if date_helper is None:
+                _date_helper = DateHelper()
+                try:
+                    import ciso8601
+                    _date_helper.parse_date = ciso8601.parse_datetime
+                except ModuleNotFoundError:
+                    _date_helper.parse_date = parser.parse
+                date_helper = _date_helper
 
     return date_helper

setup.py

@@ -26,7 +26,7 @@
     'aioresponses>=0.7.3',
     'sphinx==1.8.5',
     'sphinx_rtd_theme',
-    'jinja2==3.0.3'
+    'jinja2==3.1.2'
 ]
 
 extra_requires = [

tests/test_InfluxDBClient.py

@@ -1,3 +1,4 @@
+import codecs
 import http.server
 import json
 import logging
@@ -234,6 +235,19 @@ def test_username_password_authorization(self):
         self.client = InfluxDBClient(url=self.host, username="my-user", password="my-password", debug=True)
         self.client.query_api().query("buckets()", "my-org")
 
+    def test_query_and_debug(self):
+        self.client.close()
+        self.client = InfluxDBClient(url=self.host, token="my-token", debug=True)
+        # Query
+        results = self.client.query_api().query("buckets()", "my-org")
+        self.assertIn("my-bucket", list(map(lambda record: record["name"], results[0].records)))
+        # Query RAW
+        results = self.client.query_api().query_raw("buckets()", "my-org")
+        self.assertIn("my-bucket", codecs.decode(results.data))
+        # Bucket API
+        results = self.client.buckets_api().find_buckets()
+        self.assertIn("my-bucket", list(map(lambda bucket: bucket.name, results.buckets)))
+
     def _start_proxy_server(self):
         import http.server
         import urllib.request

tests/test_InfluxDBClientAsync.py

@@ -8,7 +8,7 @@
 import pytest
 from aioresponses import aioresponses
 
-from influxdb_client import Point, WritePrecision
+from influxdb_client import Point, WritePrecision, BucketsService
 from influxdb_client.client.exceptions import InfluxDBError
 from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
 from influxdb_client.client.warnings import MissingPivotFunction
@@ -297,6 +297,21 @@ async def test_redacted_auth_header(self):
 
         self.assertIn("Authorization: ***", log_stream.getvalue())
 
+    @async_test
+    async def test_query_and_debug(self):
+        await self.client.close()
+        self.client = InfluxDBClientAsync(url="http://localhost:8086", token="my-token", debug=True)
+        # Query
+        results = await self.client.query_api().query("buckets()", "my-org")
+        self.assertIn("my-bucket", list(map(lambda record: record["name"], results[0].records)))
+        # Query RAW
+        results = await self.client.query_api().query_raw("buckets()", "my-org")
+        self.assertIn("my-bucket", results)
+        # Bucket API
+        buckets_service = BucketsService(api_client=self.client.api_client)
+        results = await buckets_service.get_buckets()
+        self.assertIn("my-bucket", list(map(lambda bucket: bucket.name, results.buckets)))
+
     async def _prepare_data(self, measurement: str):
         _point1 = Point(measurement).tag("location", "Prague").field("temperature", 25.3)
         _point2 = Point(measurement).tag("location", "New York").field("temperature", 24.3)