This is an automated email from the ASF dual-hosted git repository. zehnder pushed a commit to branch SP-1367 in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 70ffa18001e72c4816f080e732cf4b2ba095b86a Author: Philipp Zehnder <[email protected]> AuthorDate: Wed Mar 1 10:39:04 2023 +0100 [WIP] Started to replace `DataSeries` with `QueryResult` --- .../streampipes/endpoint/api/data_lake_measure.py | 12 +- .../streampipes/model/resource/__init__.py | 4 +- .../streampipes/model/resource/data_lake_series.py | 8 +- .../streampipes/model/resource/query_result.py | 76 ++++++++++ streampipes-client-python/streampipes/test.py | 17 +++ .../tests/client/test_data_lake_series.py | 153 +++++++++------------ 6 files changed, 174 insertions(+), 96 deletions(-) diff --git a/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py b/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py index a5975b277..0c4bc6b16 100644 --- a/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py +++ b/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py @@ -26,12 +26,14 @@ from pydantic import BaseModel, Extra, Field, StrictInt, ValidationError, valida from streampipes.endpoint.endpoint import APIEndpoint from streampipes.model.container import DataLakeMeasures from streampipes.model.container.resource_container import ResourceContainer -from streampipes.model.resource import DataLakeSeries +from streampipes.model.resource import DataSeries __all__ = [ "DataLakeMeasureEndpoint", ] +from streampipes.model.resource.query_result import QueryResult + class StreamPipesQueryValidationError(Exception): """A custom exception to be raised when the validation of query parameter @@ -257,7 +259,7 @@ class DataLakeMeasureEndpoint(APIEndpoint): return config @property - def _resource_cls(self) -> Type[DataLakeSeries]: + def _resource_cls(self) -> Type[QueryResult]: """ Additional reference to resource class. This endpoint deviates from the desired relationship @@ -265,7 +267,7 @@ class DataLakeMeasureEndpoint(APIEndpoint): the return type of the get endpoint. Therefore, this is only a temporary implementation and will be removed soon. """ - return DataLakeSeries + return QueryResult @property def _container_cls(self) -> Type[ResourceContainer]: @@ -290,7 +292,7 @@ class DataLakeMeasureEndpoint(APIEndpoint): return "api", "v4", "datalake", "measurements" - def get(self, identifier: str, **kwargs: Optional[Dict[str, Any]]) -> DataLakeSeries: + def get(self, identifier: str, **kwargs: Optional[Dict[str, Any]]) -> QueryResult: """Queries the specified data lake measure from the API. By default, the maximum number of returned records is 1000. @@ -324,4 +326,4 @@ class DataLakeMeasureEndpoint(APIEndpoint): url += measurement_get_config.build_query_string() response = self._make_request(request_method=self._parent_client.request_session.get, url=url) - return self._resource_cls.from_json(json_string=response.text) + return self._resource_cls(**response.json()) diff --git a/streampipes-client-python/streampipes/model/resource/__init__.py b/streampipes-client-python/streampipes/model/resource/__init__.py index c6f0cddba..ac6c191b6 100644 --- a/streampipes-client-python/streampipes/model/resource/__init__.py +++ b/streampipes-client-python/streampipes/model/resource/__init__.py @@ -16,13 +16,13 @@ # from .data_lake_measure import DataLakeMeasure -from .data_lake_series import DataLakeSeries +from .data_lake_series import DataSeries from .data_stream import DataStream from .function_definition import FunctionDefinition __all__ = [ "DataLakeMeasure", - "DataLakeSeries", + "DataSeries", "DataStream", "FunctionDefinition", ] diff --git a/streampipes-client-python/streampipes/model/resource/data_lake_series.py b/streampipes-client-python/streampipes/model/resource/data_lake_series.py index 7d4ab8dc7..27ba3792f 100644 --- a/streampipes-client-python/streampipes/model/resource/data_lake_series.py +++ b/streampipes-client-python/streampipes/model/resource/data_lake_series.py @@ -25,7 +25,7 @@ from pydantic import StrictInt, StrictStr from streampipes.model.resource.resource import Resource __all__ = [ - "DataLakeSeries", + "DataSeries", ] @@ -41,7 +41,7 @@ class StreamPipesUnsupportedDataLakeSeries(Exception): ) -class DataLakeSeries(Resource): +class DataSeries(Resource): """Implementation of a resource for data lake series. This resource defines the data model used by its resource container(`model.container.DataLakeMeasures`). It inherits from Pydantic's BaseModel to get all its superpowers, @@ -54,7 +54,7 @@ class DataLakeSeries(Resource): """ @classmethod - def from_json(cls, json_string: str) -> DataLakeSeries: + def from_json(cls, json_string: str) -> DataSeries: """Creates an instance of `DataLakeSeries` from a given JSON string. This method is used by the resource container to parse the JSON response of @@ -68,7 +68,7 @@ class DataLakeSeries(Resource): Returns ------- - DataLakeSeries + DataSeries Instance of `DataLakeSeries` that is created based on the given JSON string. Raises diff --git a/streampipes-client-python/streampipes/model/resource/query_result.py b/streampipes-client-python/streampipes/model/resource/query_result.py new file mode 100644 index 000000000..d7b91cc50 --- /dev/null +++ b/streampipes-client-python/streampipes/model/resource/query_result.py @@ -0,0 +1,76 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from __future__ import annotations + +import json +from itertools import chain +from typing import Any, Dict, List, Optional, Union, Literal + +import pandas as pd +from pydantic import StrictInt, StrictStr, Field + +from streampipes.model.resource import DataSeries +from streampipes.model.resource.resource import Resource + +__all__ = [ + "QueryResult", +] + + +class QueryResult(Resource): + + def convert_to_pandas_representation(self) -> Dict[str, Union[List[str], List[List[Any]]]]: + """Returns the dictionary representation of a data lake series + to be used when creating a pandas Dataframe. + + It contains only the "header rows" (the column names) and "rows" that contain the actual data. + + Returns + ------- + Dictionary + Dictionary with the keys `headers` and `rows` + + """ + return self.dict(include={"headers", "rows"}) + + total: StrictInt + headers: List[StrictStr] + all_data_series: List[DataSeries] + query_status: Literal['OK', 'TOO_MUCH_DATA'] = Field(alias="spQueryStatus") + + def to_pandas(self) -> pd.DataFrame: + """Returns the data lake series in representation of a Pandas Dataframe. + + Returns + ------- + pd.DataFrame + """ + + # Pseudocode: + # pandas_representation = self.convert_to_pandas_representation() + # + # pd = pd.DataFrame(data=chain(*[item.rows for item in self.all_data_series]), + # columns=pandas_representation["headers"]) + # + # if tags not None: + # pd.groupBy(tags) + + return pd + + + test ={}) \ No newline at end of file diff --git a/streampipes-client-python/streampipes/test.py b/streampipes-client-python/streampipes/test.py new file mode 100644 index 000000000..97bceaa72 --- /dev/null +++ b/streampipes-client-python/streampipes/test.py @@ -0,0 +1,17 @@ +from streampipes.client import StreamPipesClient +from streampipes.client.config import StreamPipesClientConfig +from streampipes.client.credential_provider import StreamPipesApiKeyCredentials + +if __name__ == '__main__': + config = StreamPipesClientConfig( + credential_provider=StreamPipesApiKeyCredentials( + username="[email protected]", + api_key="Mbias0Uqdytro5fMEMnXXBYM", + ), + host_address="localhost", + https_disabled=True, + port=8082 + ) + + client = StreamPipesClient(client_config=config) + print(client.dataLakeMeasureApi.get('test').to_pandas()) diff --git a/streampipes-client-python/tests/client/test_data_lake_series.py b/streampipes-client-python/tests/client/test_data_lake_series.py index 482c0f75b..694104773 100644 --- a/streampipes-client-python/tests/client/test_data_lake_series.py +++ b/streampipes-client-python/tests/client/test_data_lake_series.py @@ -28,65 +28,8 @@ from streampipes.model.resource.data_lake_series import ( class TestDataLakeSeries(TestCase): def setUp(self) -> None: - self.series_regular = { - "total": 1, - "headers": [ - "time", - "changeDetectedHigh", - "changeDetectedLow", - "cumSumHigh", - "cumSumLow", - "level", - "overflow", - "sensorId", - "underflow", - ], - "allDataSeries": [ - { - "total": 2, - "rows": [ - [ - "2022-11-05T14:47:50.838Z", - False, - False, - "0.0", - "0.0", - 73.37740325927734, - False, - "level01", - False, - ], - [ - "2022-11-05T14:47:54.906Z", - False, - False, - "0.0", - "-0.38673634857474815", - 70.03279876708984, - False, - "level01", - False, - ], - ], - "tags": None, - "headers": [ - "time", - "changeDetectedHigh", - "changeDetectedLow", - "cumSumHigh", - "cumSumLow", - "level", - "overflow", - "sensorId", - "underflow", - ], - } - ], - } - self.series_missing = { - "total": 1, - "headers": [ + self.headers = [ "time", "changeDetectedHigh", "changeDetectedLow", @@ -96,14 +39,42 @@ class TestDataLakeSeries(TestCase): "overflow", "sensorId", "underflow", + ] + + self.data_series = { + "total": 2, + "rows": [ + [ + "2022-11-05T14:47:50.838Z", + False, + False, + "0.0", + "0.0", + 73.37740325927734, + False, + "level01", + False, + ], + [ + "2022-11-05T14:47:54.906Z", + False, + False, + "0.0", + "-0.38673634857474815", + 70.03279876708984, + False, + "level01", + False, + ], ], - "allDataSeries": [], + "tags": None, + "headers": self.headers, } - @patch("streampipes.client.client.Session", autospec=True) - def test_to_pandas(self, http_session: MagicMock): + @staticmethod + def get_result_as_panda(http_session: MagicMock, data: dict): http_session_mock = MagicMock() - http_session_mock.get.return_value.text = json.dumps(self.series_regular) + http_session_mock.get.return_value.json.return_value = data http_session.return_value = http_session_mock client = StreamPipesClient( @@ -112,6 +83,7 @@ class TestDataLakeSeries(TestCase): host_address="localhost", ) ) + result = client.dataLakeMeasureApi.get(identifier="test") http_session.assert_has_calls( @@ -119,36 +91,47 @@ class TestDataLakeSeries(TestCase): any_order=True, ) - result_pd = result.to_pandas() + return result.to_pandas() + + + @patch("streampipes.client.client.Session", autospec=True) + def test_to_pandas(self, http_session: MagicMock): + + query_result = { + "total": 1, + "headers": self.headers, + "spQueryStatus": "OK", + "allDataSeries": [ + self.data_series + ], + } + + result_pd = self.get_result_as_panda(http_session, query_result) self.assertEqual(2, len(result_pd)) self.assertListEqual( - [ - "time", - "changeDetectedHigh", - "changeDetectedLow", - "cumSumHigh", - "cumSumLow", - "level", - "overflow", - "sensorId", - "underflow", - ], + self.headers, list(result_pd.columns), ) self.assertEqual(73.37740325927734, result_pd["level"][0]) @patch("streampipes.client.client.Session", autospec=True) - def test_to_pandas_unsupported_series(self, http_session: MagicMock): - http_session_mock = MagicMock() - http_session_mock.get.return_value.text = json.dumps(self.series_missing) - http_session.return_value = http_session_mock + def test_group_by_to_pandas(self, http_session: MagicMock): + query_result = { + "total": 2, + "headers": self.headers, + "spQueryStatus": "OK", + "allDataSeries": [ + self.data_series, + self.data_series + ], + } - client = StreamPipesClient( - client_config=StreamPipesClientConfig( - credential_provider=StreamPipesApiKeyCredentials(username="user", api_key="key"), - host_address="localhost", - ) + result_pd = self.get_result_as_panda(http_session, query_result) + + self.assertEqual(4, len(result_pd)) + self.assertListEqual( + self.headers, + list(result_pd.columns), ) - with self.assertRaises(StreamPipesUnsupportedDataLakeSeries): - client.dataLakeMeasureApi.get(identifier="test") + self.assertEqual(70.03279876708984, result_pd["level"][3])
