feat: CSV parser is able to parse export from UI (#171)

Author: bednar

CHANGELOG.md

@@ -1,8 +1,10 @@
 ## 1.13.0 [unreleased]
 
-### Bug Fixes
-1. [#164](https://github.com/influxdata/influxdb-client-python/pull/170): Skip DataFrame rows without data - all fields are nan. 
+### Features
+1. [#171](https://github.com/influxdata/influxdb-client-python/pull/171): CSV parser is able to parse export from UI
 
+### Bug Fixes
+1. [#170](https://github.com/influxdata/influxdb-client-python/pull/170): Skip DataFrame rows without data - all fields are nan. 
 
 ## 1.12.0 [2020-10-30]
 

influxdb_client/client/flux_csv_parser.py

@@ -13,6 +13,12 @@
 from influxdb_client.client.flux_table import FluxTable, FluxColumn, FluxRecord
 
 
+ANNOTATION_DEFAULT = "#default"
+ANNOTATION_GROUP = "#group"
+ANNOTATION_DATATYPE = "#datatype"
+ANNOTATIONS = [ANNOTATION_DEFAULT, ANNOTATION_GROUP, ANNOTATION_DATATYPE]
+
+
 class FluxQueryException(Exception):
     """The exception from InfluxDB."""
 
@@ -68,6 +74,7 @@ def _parse_flux_response(self):
         table_id = -1
         start_new_table = False
         table = None
+        groups = []
         parsing_state_error = False
 
         for csv in self._reader:
@@ -90,7 +97,7 @@ def _parse_flux_response(self):
 
             token = csv[0]
             # start    new    table
-            if "#datatype" == token:
+            if token in ANNOTATIONS and not start_new_table:
 
                 # Return already parsed DataFrame
                 if (self._serialization_mode is FluxSerializationMode.dataFrame) & hasattr(self, '_data_frame'):
@@ -105,18 +112,19 @@ def _parse_flux_response(self):
                 raise FluxCsvParserException("Unable to parse CSV response. FluxTable definition was not found.")
 
             #  # datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,dateTime:RFC3339,double,string,string,string
-            if "#datatype" == token:
+            if ANNOTATION_DATATYPE == token:
                 self.add_data_types(table, csv)
 
-            elif "#group" == token:
-                self.add_groups(table, csv)
+            elif ANNOTATION_GROUP == token:
+                groups = csv
 
-            elif "#default" == token:
+            elif ANNOTATION_DEFAULT == token:
                 self.add_default_empty_values(table, csv)
 
             else:
                 # parse column names
                 if start_new_table:
+                    self.add_groups(table, groups)
                     self.add_column_names_and_tags(table, csv)
                     start_new_table = False
                     # Create DataFrame with default values

tests/test_FluxCSVParser.py

@@ -141,6 +141,23 @@ def test_response_with_error(self):
         self.assertEqual('engine: unknown field type for value: xyz', exception.message)
         self.assertEqual('', exception.reference)
 
+    def test_ParseExportFromUserInterface(self):
+
+        data = "#group,false,false,true,true,true,true,true,true,false,false\n" \
+           + "#datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,string,string,string,string,double,dateTime:RFC3339\n" \
+           + "#default,mean,,,,,,,,,\n" \
+           + ",result,table,_start,_stop,_field,_measurement,city,location,_value,_time\n" \
+           + ",,0,1754-06-26T11:30:27.613654848Z,2040-10-27T12:13:46.485Z,temperatureC,weather,London,us-midwest,30,1975-09-01T16:59:54.5Z\n" \
+           + ",,1,1754-06-26T11:30:27.613654848Z,2040-10-27T12:13:46.485Z,temperatureF,weather,London,us-midwest,86,1975-09-01T16:59:54.5Z\n";
+
+        tables = self._parse_to_tables(data=data)
+        self.assertEqual(2, tables.__len__())
+        self.assertEqual(1, tables[0].records.__len__())
+        self.assertEqual(1, tables[1].records.__len__())
+        self.assertFalse(tables[1].columns[0].group)
+        self.assertFalse(tables[1].columns[1].group)
+        self.assertTrue(tables[1].columns[2].group)
+
     @staticmethod
     def _parse_to_tables(data: str):
         fp = BytesIO(str.encode(data))