fix(async): parsing query response with two-bytes UTF-8 character

CHANGELOG.md

@@ -5,6 +5,7 @@
 
 ### Bug Fixes
 1. [#512](https://github.com/influxdata/influxdb-client-python/pull/512): Exception propagation for asynchronous `QueryApi` [async/await]
+1. [#518](https://github.com/influxdata/influxdb-client-python/pull/518): Parsing query response with two-bytes UTF-8 character [async/await]
 
 ## 1.33.0 [2022-09-29]
 

influxdb_client/client/flux_csv_parser.py

@@ -384,6 +384,10 @@ def _print_profiler_info(self, flux_record: FluxRecord):
 class _StreamReaderToWithAsyncRead:
     def __init__(self, response):
         self.response = response
+        self.decoder = codecs.getincrementaldecoder(_UTF_8_encoding)()
 
     async def read(self, size: int) -> str:
-        return (await self.response.read(size)).decode(_UTF_8_encoding)
+        raw_bytes = (await self.response.read(size))
+        if not raw_bytes:
+            return self.decoder.decode(b'', final=True)
+        return self.decoder.decode(raw_bytes, final=False)

tests/test_InfluxDBClientAsync.py

@@ -383,6 +383,25 @@ async def test_query_exception_propagation(self):
             await self.client.query_api().query("buckets()", "my-org")
         self.assertEqual("unauthorized access", e.value.message)
 
+    @async_test
+    @aioresponses()
+    async def test_parse_utf8_two_bytes_character(self, mocked):
+        await self.client.close()
+        self.client = InfluxDBClientAsync("http://localhost")
+
+        body = '''#group,false,false,false,false,true,true,true
+#datatype,string,long,dateTime:RFC3339,string,string,string,string
+#default,_result,,,,,,
+,result,table,_time,_value,_field,_measurement,type
+'''
+        for i in range(1000):
+            body += f",,0,2022-10-13T12:28:31.{i}Z,ÂÂÂ,value,async,error\n"
+
+        mocked.post('http://localhost/api/v2/query?org=my-org', status=200, body=body)
+
+        data_frame = await self.client.query_api().query_data_frame("from()", "my-org")
+        self.assertEqual(1000, len(data_frame))
+
     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)