feat(query): improve Query UX (influxdata#450)

Author: bednar

CHANGELOG.md

@@ -10,6 +10,7 @@
 
 ### Features
 1. [#440](https://github.com/influxdata/influxdb-client-python/pull/440): Add possibility to specify timestamp column and its timezone [DataFrame]
+1. [#450](https://github.com/influxdata/influxdb-client-python/pull/450): Improve Query UX - simplify serialization to JSON and add possibility to serialize query results as a flattened list of values
 
 ### Bug Fixes
 1. [#457](https://github.com/influxdata/influxdb-client-python/pull/457): Formatting nanoseconds to Flux AST

README.rst

@@ -615,10 +615,9 @@ Queries
 The result retrieved by `QueryApi <https://github.com/influxdata/influxdb-client-python/blob/master/influxdb_client/client/query_api.py>`_  could be formatted as a:
 
 1. Flux data structure: `FluxTable <https://github.com/influxdata/influxdb-client-python/blob/master/influxdb_client/client/flux_table.py#L5>`_, `FluxColumn <https://github.com/influxdata/influxdb-client-python/blob/master/influxdb_client/client/flux_table.py#L22>`_ and `FluxRecord <https://github.com/influxdata/influxdb-client-python/blob/master/influxdb_client/client/flux_table.py#L31>`_
-2. Query bind parameters
-3. `csv.reader <https://docs.python.org/3.4/library/csv.html#reader-objects>`__ which will iterate over CSV lines
-4. Raw unprocessed results as a ``str`` iterator
-5. `Pandas DataFrame <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html>`_
+2. :class:`~influxdb_client.client.flux_table.CSVIterator` which will iterate over CSV lines
+3. Raw unprocessed results as a ``str`` iterator
+4. `Pandas DataFrame <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html>`_
 
 The API also support streaming ``FluxRecord`` via `query_stream <https://github.com/influxdata/influxdb-client-python/blob/master/influxdb_client/client/query_api.py#L77>`_, see example below:
 

docs/api.rst

@@ -20,6 +20,12 @@ QueryApi
 .. autoclass:: influxdb_client.client.flux_table.FluxRecord
    :members:
 
+.. autoclass:: influxdb_client.client.flux_table.TableList
+   :members:
+
+.. autoclass:: influxdb_client.client.flux_table.CSVIterator
+   :members:
+
 WriteApi
 """"""""
 .. autoclass:: influxdb_client.WriteApi

examples/query_response_to_json.py

@@ -19,8 +19,5 @@
     """
     Serialize to JSON
     """
-    import json
-    from influxdb_client.client.flux_table import FluxStructureEncoder
-
-    output = json.dumps(tables, cls=FluxStructureEncoder, indent=2)
+    output = tables.to_json(indent=5)
     print(output)

influxdb_client/client/_base.py

@@ -2,13 +2,11 @@
 from __future__ import absolute_import
 
 import base64
-import codecs
 import configparser
-import csv
 import logging
 import os
 from datetime import datetime, timedelta
-from typing import Iterator, List, Generator, Any, Union, Iterable, AsyncGenerator
+from typing import List, Generator, Any, Union, Iterable, AsyncGenerator
 
 from urllib3 import HTTPResponse
 
@@ -17,9 +15,10 @@
     Duration, StringLiteral, ArrayExpression, ImportDeclaration, MemberExpression, MemberAssignment, File, \
     WriteService, QueryService, DeleteService, DeletePredicateRequest
 from influxdb_client.client.flux_csv_parser import FluxResponseMetadataMode, FluxCsvParser, FluxSerializationMode
-from influxdb_client.client.flux_table import FluxTable, FluxRecord
+from influxdb_client.client.flux_table import FluxRecord, TableList, CSVIterator
 from influxdb_client.client.util.date_utils import get_date_helper
 from influxdb_client.client.util.helpers import get_org_query_param
+from influxdb_client.client.warnings import MissingPivotFunction
 from influxdb_client.client.write.dataframe_serializer import DataframeSerializer
 from influxdb_client.rest import _UTF_8_encoding
 
@@ -214,9 +213,9 @@ def __init__(self, influxdb_client, query_options=None):
     """Base implementation for Queryable API."""
 
     def _to_tables(self, response, query_options=None, response_metadata_mode:
-                   FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> List[FluxTable]:
+                   FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> TableList:
         """
-        Parse HTTP response to FluxTables.
+        Parse HTTP response to TableList.
 
         :param response: HTTP response from an HTTP client. Expected type: `urllib3.response.HTTPResponse`.
         """
@@ -225,9 +224,9 @@ def _to_tables(self, response, query_options=None, response_metadata_mode:
         return _parser.table_list()
 
     async def _to_tables_async(self, response, query_options=None, response_metadata_mode:
-                               FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> List[FluxTable]:
+                               FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> TableList:
         """
-        Parse HTTP response to FluxTables.
+        Parse HTTP response to TableList.
 
         :param response: HTTP response from an HTTP client. Expected type: `aiohttp.client_reqrep.ClientResponse`.
         """
@@ -236,9 +235,9 @@ async def _to_tables_async(self, response, query_options=None, response_metadata
                 pass
             return parser.table_list()
 
-    def _to_csv(self, response: HTTPResponse) -> Iterator[List[str]]:
+    def _to_csv(self, response: HTTPResponse) -> CSVIterator:
         """Parse HTTP response to CSV."""
-        return csv.reader(codecs.iterdecode(response, _UTF_8_encoding))
+        return CSVIterator(response)
 
     def _to_flux_record_stream(self, response, query_options=None,
                                response_metadata_mode: FluxResponseMetadataMode = FluxResponseMetadataMode.full) -> \
@@ -320,7 +319,7 @@ def _get_query_options(self):
             from influxdb_client.client.query_api import QueryOptions
             return QueryOptions(profilers=self._influxdb_client.profilers)
 
-    def _create_query(self, query, dialect=default_dialect, params: dict = None):
+    def _create_query(self, query, dialect=default_dialect, params: dict = None, **kwargs):
         query_options = self._get_query_options()
         profilers = query_options.profilers if query_options is not None else None
         q = Query(query=query, dialect=dialect, extern=_BaseQueryApi._build_flux_ast(params, profilers))
@@ -331,6 +330,9 @@ def _create_query(self, query, dialect=default_dialect, params: dict = None):
             print("===============")
             print(query)
 
+        if kwargs.get('dataframe_query', False):
+            MissingPivotFunction.print_warning(query)
+
         return q
 
     @staticmethod

influxdb_client/client/flux_csv_parser.py

@@ -7,7 +7,7 @@
 from enum import Enum
 from typing import List
 
-from influxdb_client.client.flux_table import FluxTable, FluxColumn, FluxRecord
+from influxdb_client.client.flux_table import FluxTable, FluxColumn, FluxRecord, TableList
 from influxdb_client.client.util.date_utils import get_date_helper
 from influxdb_client.rest import _UTF_8_encoding
 
@@ -71,7 +71,7 @@ def __init__(self, response, serialization_mode: FluxSerializationMode,
                          Acceptable types: `urllib3.response.HTTPResponse`, `aiohttp.client_reqrep.ClientResponse`.
         """
         self._response = response
-        self.tables = []
+        self.tables = TableList()
         self._serialization_mode = serialization_mode
         self._response_metadata_mode = response_metadata_mode
         self._data_frame_index = data_frame_index
@@ -344,12 +344,12 @@ def _is_profiler_table(self, table: FluxTable) -> bool:
         return any(filter(lambda column: (column.default_value == "_profiler" and column.label == "result"),
                           table.columns))
 
-    def table_list(self) -> List[FluxTable]:
+    def table_list(self) -> TableList:
         """Get the list of flux tables."""
         if not self._profilers:
             return self.tables
         else:
-            return list(filter(lambda table: not self._is_profiler_table(table), self.tables))
+            return TableList(filter(lambda table: not self._is_profiler_table(table), self.tables))
 
     def _print_profiler_info(self, flux_record: FluxRecord):
         if flux_record.get_measurement().startswith("profiler/"):

influxdb_client/client/flux_table.py

@@ -3,7 +3,12 @@
 
 The data model consists of tables, records, columns.
 """
+import codecs
+import csv
+from http.client import HTTPResponse
 from json import JSONEncoder
+from typing import List, Iterator
+from influxdb_client.rest import _UTF_8_encoding
 
 
 class FluxStructure:
@@ -137,3 +142,145 @@ def __str__(self):
     def __repr__(self):
         """Format for inspection."""
         return f"<{type(self).__name__}: field={self.values.get('_field')}, value={self.values.get('_value')}>"
+
+
+class TableList(List[FluxTable]):
+    """:class:`~influxdb_client.client.flux_table.FluxTable` list with additionally functional to better handle of query result."""  # noqa: E501
+
+    def to_values(self, columns: List['str'] = None) -> List[List[object]]:
+        """
+        Serialize query results to a flattened list of values.
+
+        :param columns: if not ``None`` then only specified columns are presented in results
+        :return: :class:`~list` of values
+
+        Output example:
+
+        .. code-block:: python
+
+            [
+                ['New York', datetime.datetime(2022, 6, 7, 11, 3, 22, 917593, tzinfo=tzutc()), 24.3],
+                ['Prague', datetime.datetime(2022, 6, 7, 11, 3, 22, 917593, tzinfo=tzutc()), 25.3],
+                ...
+            ]
+
+        Configure required columns:
+
+        .. code-block:: python
+
+            from influxdb_client import InfluxDBClient
+
+                with InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") as client:
+
+                # Query: using Table structure
+                tables = client.query_api().query('from(bucket:"my-bucket") |> range(start: -10m)')
+
+                # Serialize to values
+                output = tables.to_values(columns=['location', '_time', '_value'])
+                print(output)
+        """
+
+        def filter_values(record):
+            if columns is not None:
+                return [record.values.get(k) for k in columns]
+            return record.values.values()
+
+        return self._to_values(filter_values)
+
+    def to_json(self, columns: List['str'] = None, **kwargs) -> str:
+        """
+        Serialize query results to a JSON formatted :class:`~str`.
+
+        :param columns: if not ``None`` then only specified columns are presented in results
+        :return: :class:`~str`
+
+        The query results is flattened to array:
+
+        .. code-block:: javascript
+
+            [
+                {
+                    "_measurement": "mem",
+                    "_start": "2021-06-23T06:50:11.897825+00:00",
+                    "_stop": "2021-06-25T06:50:11.897825+00:00",
+                    "_time": "2020-02-27T16:20:00.897825+00:00",
+                    "region": "north",
+                     "_field": "usage",
+                    "_value": 15
+                },
+                {
+                    "_measurement": "mem",
+                    "_start": "2021-06-23T06:50:11.897825+00:00",
+                    "_stop": "2021-06-25T06:50:11.897825+00:00",
+                    "_time": "2020-02-27T16:20:01.897825+00:00",
+                    "region": "west",
+                     "_field": "usage",
+                    "_value": 10
+                },
+                ...
+            ]
+
+        The JSON format could be configured via ``**kwargs`` arguments:
+
+        .. code-block:: python
+
+            from influxdb_client import InfluxDBClient
+
+            with InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") as client:
+
+                # Query: using Table structure
+                tables = client.query_api().query('from(bucket:"my-bucket") |> range(start: -10m)')
+
+                # Serialize to JSON
+                output = tables.to_json(indent=5)
+                print(output)
+
+        For all available options see - `json.dump <https://docs.python.org/3/library/json.html#json.dump>`_.
+        """
+        if 'indent' not in kwargs:
+            kwargs['indent'] = 2
+
+        def filter_values(record):
+            if columns is not None:
+                return {k: v for (k, v) in record.values.items() if k in columns}
+            return record.values
+
+        import json
+        return json.dumps(self._to_values(filter_values), cls=FluxStructureEncoder, **kwargs)
+
+    def _to_values(self, mapping):
+        return [mapping(record) for table in self for record in table.records]
+
+
+class CSVIterator(Iterator[List[str]]):
+    """:class:`Iterator[List[str]]` with additionally functional to better handle of query result."""
+
+    def __init__(self, response: HTTPResponse) -> None:
+        """Initialize ``csv.reader``."""
+        self.delegate = csv.reader(codecs.iterdecode(response, _UTF_8_encoding))
+
+    def __iter__(self):
+        """Return an iterator object."""
+        return self.delegate.__iter__()
+
+    def __next__(self):
+        """Retrieve the next item from the iterator."""
+        return self.delegate.__next__()
+
+    def to_values(self) -> List[List[str]]:
+        """
+        Serialize query results to a flattened list of values.
+
+        :return: :class:`~list` of values
+
+        Output example:
+
+        .. code-block:: python
+
+            [
+                ['New York', '2022-06-14T08:00:51.749072045Z', '24.3'],
+                ['Prague', '2022-06-14T08:00:51.749072045Z', '25.3'],
+                ...
+            ]
+        """
+        return list(self.delegate)