fix: parsing response with new line in column [async/await]

Author: bednar

README.rst

@@ -1371,12 +1371,12 @@ How to use Asyncio
 .. marker-asyncio-start
 
 Starting from version 1.27.0 for Python 3.7+ the ``influxdb-client`` package supports ``async/await`` based on
-`asyncio <https://docs.python.org/3/library/asyncio.html>`_ and `aiohttp <https://docs.aiohttp.org>`_.
-You can install ``aiohttp`` directly:
+`asyncio <https://docs.python.org/3/library/asyncio.html>`_, `aiohttp <https://docs.aiohttp.org>`_ and `aiocsv <https://pypi.org/project/aiocsv/>`_.
+You can install ``aiohttp`` and ``aiocsv`` directly:
 
  .. code-block:: bash
 
-    $ python -m pip install influxdb-client aiohttp
+    $ python -m pip install influxdb-client aiohttp aiocsv
 
 or use the ``[async]`` extra:
 

influxdb_client/client/flux_csv_parser.py

@@ -99,7 +99,8 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     async def __aenter__(self) -> 'FluxCsvParser':
         """Initialize CSV reader."""
-        self._reader = self._response.content
+        from aiocsv import AsyncReader
+        self._reader = AsyncReader(_StreamReaderToWithAsyncRead(self._response.content))
 
         return self
 
@@ -134,11 +135,9 @@ async def _parse_flux_response_async(self):
         metadata = _FluxCsvParserMetadata()
 
         try:
-            async for line in self._reader:
-                csv = list(csv_parser.reader([line.decode(_UTF_8_encoding)]))
-                if len(csv) >= 1:
-                    for val in self._parse_flux_response_row(metadata, csv[0]):
-                        yield val
+            async for csv in self._reader:
+                for val in self._parse_flux_response_row(metadata, csv):
+                    yield val
 
             # Return latest DataFrame
             if (self._serialization_mode is FluxSerializationMode.dataFrame) & hasattr(self, '_data_frame'):
@@ -371,3 +370,11 @@ def _print_profiler_info(self, flux_record: FluxRecord):
                         print(f"{name:<20}: \n\n{val}")
                     elif val is not None:
                         print(f"{name:<20}: {val:<20}")
+
+
+class _StreamReaderToWithAsyncRead:
+    def __init__(self, response):
+        self.response = response
+
+    async def read(self, size: int) -> str:
+        return (await self.response.read(size)).decode(_UTF_8_encoding)

setup.py

@@ -39,7 +39,8 @@
 ]
 
 async_requires = [
-    'aiohttp>=3.8.1'
+    'aiohttp>=3.8.1',
+    'aiocsv>=1.2.2'
 ]
 
 with open('README.rst', 'r') as f: