You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by bo...@apache.org on 2022/11/24 19:02:21 UTC
[streampipes] branch STREAMPIPES-607 created (now bd47fda94)
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a change to branch STREAMPIPES-607
in repository https://gitbox.apache.org/repos/asf/streampipes.git
at bd47fda94 Merge remote-tracking branch 'bossenti/STREAMPIPES-607' into STREAMPIPES-607
This branch includes the following new commits:
new cc4f870a5 experimental "get()"
new 01a66127c experimental "get()"
new fe6ff0f2a experimental get()
new bd47fda94 Merge remote-tracking branch 'bossenti/STREAMPIPES-607' into STREAMPIPES-607
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[streampipes] 01/04: experimental "get()"
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch STREAMPIPES-607
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit cc4f870a57c8c417e55f7d3ee497c3e475091fc9
Author: bossenti <bo...@posteo.de>
AuthorDate: Mon Nov 14 20:40:55 2022 +0100
experimental "get()"
---
.../endpoint/data_lake_measure.py | 27 +++++++++-
.../model/resource/DataLakeSeries.py | 58 ++++++++++++++++++++++
2 files changed, 83 insertions(+), 2 deletions(-)
diff --git a/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py b/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
index 168c14728..d854b60b4 100644
--- a/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
+++ b/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
@@ -23,13 +23,14 @@ from typing import Tuple, Type
from streampipes_client.endpoint.endpoint import APIEndpoint
from streampipes_client.model.container import DataLakeMeasures
+from streampipes_client.model.container.resource_container import ResourceContainer
+from streampipes_client.model.resource.DataLakeSeries import DataLakeSeries
+from streampipes_client.model.resource.resource import Resource
__all__ = [
"DataLakeMeasureEndpoint",
]
-from streampipes_client.model.container.resource_container import ResourceContainer
-
class DataLakeMeasureEndpoint(APIEndpoint):
"""Implementation of the DataLakeMeasure endpoint.
@@ -68,6 +69,10 @@ class DataLakeMeasureEndpoint(APIEndpoint):
5
"""
+ @property
+ def _resource_cls(self) -> Type[DataLakeSeries]:
+ return DataLakeSeries
+
@property
def _container_cls(self) -> Type[ResourceContainer]:
"""Defines the model container class the endpoint refers to.
@@ -90,3 +95,21 @@ class DataLakeMeasureEndpoint(APIEndpoint):
"""
return "api", "v4", "datalake", "measurements"
+
+ def get(self, identifier: str) -> Resource:
+ """Queries the specified resource from the API endpoint.
+
+ Parameters
+ ----------
+ identifier: str
+ The identifier of the resource to be queried.
+
+ Returns
+ -------
+ The specified resource as an instance of the corresponding model class (`model.Element`).
+ """
+
+ response = self._make_request(
+ request_method=self._parent_client.request_session.get, url=f"{self.build_url()}/{identifier}"
+ )
+ return self._resource_cls.from_json(json_string=response.text)
diff --git a/streampipes-client-python/streampipes_client/model/resource/DataLakeSeries.py b/streampipes-client-python/streampipes_client/model/resource/DataLakeSeries.py
new file mode 100644
index 000000000..a936edafc
--- /dev/null
+++ b/streampipes-client-python/streampipes_client/model/resource/DataLakeSeries.py
@@ -0,0 +1,58 @@
+#
+# 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.
+#
+import json
+from typing import Any, Dict, List, Optional
+
+import pandas as pd
+from pydantic import StrictInt, StrictStr
+from streampipes_client.model.resource.resource import Resource
+
+
+class DataLakeSeries(Resource):
+ @classmethod
+ def from_json(cls, json_string: str) -> Resource:
+
+ # deserialize JSON string
+ parsed_json = json.loads(json_string)
+
+ if len(parsed_json["allDataSeries"]) != 1:
+ raise RuntimeError("Not supported")
+
+ data_series = parsed_json["allDataSeries"][0]
+
+ return cls.parse_obj(data_series)
+
+ def convert_to_pandas_representation(self) -> Dict:
+
+ result: Dict = dict()
+
+ for row in self.rows:
+ for idx, value in enumerate(row):
+ if (key := self.headers[idx]) in result.keys():
+ result[self.headers[idx]].append(value)
+ else:
+ result.update({key: [value]})
+
+ return result
+
+ total: StrictInt
+ headers: List[StrictStr]
+ rows: List[List[Any]]
+ tags: Optional[str]
+
+ def to_pandas(self) -> pd.DataFrame:
+ return pd.DataFrame(data=self.convert_to_pandas_representation())
[streampipes] 02/04: experimental "get()"
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch STREAMPIPES-607
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 01a66127cf77927d82fc032a60855e1664ff9a83
Author: bossenti <bo...@posteo.de>
AuthorDate: Mon Nov 14 20:40:55 2022 +0100
experimental "get()"
---
.../endpoint/data_lake_measure.py | 27 +++++++++-
.../model/resource/DataLakeSeries.py | 58 ++++++++++++++++++++++
2 files changed, 83 insertions(+), 2 deletions(-)
diff --git a/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py b/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
index 168c14728..d854b60b4 100644
--- a/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
+++ b/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
@@ -23,13 +23,14 @@ from typing import Tuple, Type
from streampipes_client.endpoint.endpoint import APIEndpoint
from streampipes_client.model.container import DataLakeMeasures
+from streampipes_client.model.container.resource_container import ResourceContainer
+from streampipes_client.model.resource.DataLakeSeries import DataLakeSeries
+from streampipes_client.model.resource.resource import Resource
__all__ = [
"DataLakeMeasureEndpoint",
]
-from streampipes_client.model.container.resource_container import ResourceContainer
-
class DataLakeMeasureEndpoint(APIEndpoint):
"""Implementation of the DataLakeMeasure endpoint.
@@ -68,6 +69,10 @@ class DataLakeMeasureEndpoint(APIEndpoint):
5
"""
+ @property
+ def _resource_cls(self) -> Type[DataLakeSeries]:
+ return DataLakeSeries
+
@property
def _container_cls(self) -> Type[ResourceContainer]:
"""Defines the model container class the endpoint refers to.
@@ -90,3 +95,21 @@ class DataLakeMeasureEndpoint(APIEndpoint):
"""
return "api", "v4", "datalake", "measurements"
+
+ def get(self, identifier: str) -> Resource:
+ """Queries the specified resource from the API endpoint.
+
+ Parameters
+ ----------
+ identifier: str
+ The identifier of the resource to be queried.
+
+ Returns
+ -------
+ The specified resource as an instance of the corresponding model class (`model.Element`).
+ """
+
+ response = self._make_request(
+ request_method=self._parent_client.request_session.get, url=f"{self.build_url()}/{identifier}"
+ )
+ return self._resource_cls.from_json(json_string=response.text)
diff --git a/streampipes-client-python/streampipes_client/model/resource/DataLakeSeries.py b/streampipes-client-python/streampipes_client/model/resource/DataLakeSeries.py
new file mode 100644
index 000000000..a936edafc
--- /dev/null
+++ b/streampipes-client-python/streampipes_client/model/resource/DataLakeSeries.py
@@ -0,0 +1,58 @@
+#
+# 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.
+#
+import json
+from typing import Any, Dict, List, Optional
+
+import pandas as pd
+from pydantic import StrictInt, StrictStr
+from streampipes_client.model.resource.resource import Resource
+
+
+class DataLakeSeries(Resource):
+ @classmethod
+ def from_json(cls, json_string: str) -> Resource:
+
+ # deserialize JSON string
+ parsed_json = json.loads(json_string)
+
+ if len(parsed_json["allDataSeries"]) != 1:
+ raise RuntimeError("Not supported")
+
+ data_series = parsed_json["allDataSeries"][0]
+
+ return cls.parse_obj(data_series)
+
+ def convert_to_pandas_representation(self) -> Dict:
+
+ result: Dict = dict()
+
+ for row in self.rows:
+ for idx, value in enumerate(row):
+ if (key := self.headers[idx]) in result.keys():
+ result[self.headers[idx]].append(value)
+ else:
+ result.update({key: [value]})
+
+ return result
+
+ total: StrictInt
+ headers: List[StrictStr]
+ rows: List[List[Any]]
+ tags: Optional[str]
+
+ def to_pandas(self) -> pd.DataFrame:
+ return pd.DataFrame(data=self.convert_to_pandas_representation())
[streampipes] 03/04: experimental get()
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch STREAMPIPES-607
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit fe6ff0f2a129cfca3838873b98511a98c05de8c0
Author: bossenti <bo...@posteo.de>
AuthorDate: Sat Nov 19 09:10:42 2022 +0100
experimental get()
---
streampipes-client-python/.pre-commit-config.yaml | 2 +-
.../streampipes_client/client/client_config.py | 5 +-
.../endpoint/data_lake_measure.py | 18 ++-
.../streampipes_client/endpoint/endpoint.py | 2 +-
.../model/container/resource_container.py | 2 +-
.../model/resource/data_lake_series.py | 118 ++++++++++++++++
.../tests/client/test_client.py | 9 +-
.../tests/client/test_data_lake_series.py | 154 +++++++++++++++++++++
8 files changed, 295 insertions(+), 15 deletions(-)
diff --git a/streampipes-client-python/.pre-commit-config.yaml b/streampipes-client-python/.pre-commit-config.yaml
index d353a24d7..ecc2aec76 100644
--- a/streampipes-client-python/.pre-commit-config.yaml
+++ b/streampipes-client-python/.pre-commit-config.yaml
@@ -33,7 +33,7 @@ repos:
name: interrogate
language: python
types: [ python ]
- entry: interrogate -vv --fail-under 95 --omit-covered-files --ignore-init-method --ignore-module --ignore-magic
+ entry: interrogate -vv --fail-under 95 --omit-covered-files --ignore-init-method --ignore-module --ignore-magic --ignore-regex test_* --ignore-regex Test*
- id: pyupgrade
name: pyupgrade
diff --git a/streampipes-client-python/streampipes_client/client/client_config.py b/streampipes-client-python/streampipes_client/client/client_config.py
index 9abcbdaf0..0e2a38144 100644
--- a/streampipes-client-python/streampipes_client/client/client_config.py
+++ b/streampipes-client-python/streampipes_client/client/client_config.py
@@ -27,6 +27,7 @@ __all__ = [
"StreamPipesClientConfig",
]
+
from streampipes_client.client.credential_provider import CredentialProvider
@@ -45,7 +46,7 @@ class StreamPipesClientConfig:
https_disabled: Optional[bool]
Determines whether https is used to connect to StreamPipes.
port: Optional[int]
- Specifies the port under which the StreamPipes API is available, e.g., 80.
+ Specifies the port under which the StreamPipes API is available, e.g., 80 (with http) or 443 (with https)
Examples
--------
@@ -55,4 +56,4 @@ class StreamPipesClientConfig:
credential_provider: CredentialProvider
host_address: str
https_disabled: Optional[bool] = False
- port: Optional[int] = 80 if https_disabled else 443
+ port: Optional[int] = 80
diff --git a/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py b/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
index d854b60b4..e2effde8d 100644
--- a/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
+++ b/streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
@@ -24,8 +24,7 @@ from typing import Tuple, Type
from streampipes_client.endpoint.endpoint import APIEndpoint
from streampipes_client.model.container import DataLakeMeasures
from streampipes_client.model.container.resource_container import ResourceContainer
-from streampipes_client.model.resource.DataLakeSeries import DataLakeSeries
-from streampipes_client.model.resource.resource import Resource
+from streampipes_client.model.resource.data_lake_series import DataLakeSeries
__all__ = [
"DataLakeMeasureEndpoint",
@@ -71,6 +70,13 @@ class DataLakeMeasureEndpoint(APIEndpoint):
@property
def _resource_cls(self) -> Type[DataLakeSeries]:
+ """
+ Additional reference to resource class.
+ This endpoint deviates from the desired relationship
+ that the resource class of the resource container is
+ the return type of the get endpoint.
+ Therefore, this is only a temporary implementation and will be removed soon.
+ """
return DataLakeSeries
@property
@@ -96,17 +102,17 @@ class DataLakeMeasureEndpoint(APIEndpoint):
return "api", "v4", "datalake", "measurements"
- def get(self, identifier: str) -> Resource:
- """Queries the specified resource from the API endpoint.
+ def get(self, identifier: str) -> DataLakeSeries:
+ """Queries the specified data lake measure from the API.
Parameters
----------
identifier: str
- The identifier of the resource to be queried.
+ The identifier of the data lake measure to be queried.
Returns
-------
- The specified resource as an instance of the corresponding model class (`model.Element`).
+ The specified data lake measure as an instance of the corresponding model class (`model.DataLakeSeries`).
"""
response = self._make_request(
diff --git a/streampipes-client-python/streampipes_client/endpoint/endpoint.py b/streampipes-client-python/streampipes_client/endpoint/endpoint.py
index 0d141fe49..b28f5e35e 100644
--- a/streampipes-client-python/streampipes_client/endpoint/endpoint.py
+++ b/streampipes-client-python/streampipes_client/endpoint/endpoint.py
@@ -189,7 +189,7 @@ class APIEndpoint(ABC):
Returns
-------
- The specified resource as an instance of the corresponding model class (`model.Element`).
+ The specified resource as an instance of the corresponding model class (`model.Resource`).
"""
raise NotImplementedError(
"We're sorry! This functionality is not yet part of the StreamPipes Python client."
diff --git a/streampipes-client-python/streampipes_client/model/container/resource_container.py b/streampipes-client-python/streampipes_client/model/container/resource_container.py
index 0b16c430e..982e6e794 100644
--- a/streampipes-client-python/streampipes_client/model/container/resource_container.py
+++ b/streampipes-client-python/streampipes_client/model/container/resource_container.py
@@ -179,8 +179,8 @@ class ResourceContainer(ABC):
# raise an exception if the response does not be a list
if not type(parsed_json) == list:
raise StreamPipesResourceContainerJSONError(container_name=str(cls), json_string=json_string)
- try:
+ try:
resource_container = cls(resources=[cls._resource_cls().parse_obj(item) for item in parsed_json])
except ValidationError as ve:
raise StreamPipesDataModelError(validation_error=ve)
diff --git a/streampipes-client-python/streampipes_client/model/resource/data_lake_series.py b/streampipes-client-python/streampipes_client/model/resource/data_lake_series.py
new file mode 100644
index 000000000..e211361a8
--- /dev/null
+++ b/streampipes-client-python/streampipes_client/model/resource/data_lake_series.py
@@ -0,0 +1,118 @@
+#
+# 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 typing import Any, Dict, List, Optional, Union
+
+import pandas as pd
+from pydantic import StrictInt, StrictStr
+from streampipes_client.model.resource.resource import Resource
+
+
+class StreamPipesUnsupportedDataLakeSeries(Exception):
+ """Exception to be raised when the returned data lake series
+ cannot be parsed with the current implementation of the resource.
+ """
+
+ def __init__(self):
+ super().__init__(
+ "The Data Lake series returned by the API appears "
+ "to have a structure that is not currently supported by the Python client."
+ )
+
+
+class DataLakeSeries(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,
+ which are used to parse, validate the API response and to easily switch between
+ the Python representation (both serialized and deserialized) and Java representation (serialized only).
+
+ NOTE:
+ This class will only exist temporarily it its current appearance since
+ there are some inconsistencies in the StreamPipes API.
+ """
+
+ @classmethod
+ def from_json(cls, json_string: str) -> DataLakeSeries:
+ """Creates an instance of `DataLakeSeries` from a given JSON string.
+
+ This method is used by the resource container to parse the JSON response of
+ the StreamPipes API.
+ Currently, it only supports data lake series that consist of exactly one series of data.
+
+ Parameters
+ ----------
+ json_string: str
+ The JSON string the data lake series should be created on.
+
+ Returns
+ -------
+ DataLakeSeries
+ Instance of `DataLakeSeries` that is created based on the given JSON string.
+
+ Raises
+ ------
+ StreamPipesUnsupportedDataLakeSeries
+ If the data lake series returned by the StreamPipes API cannot be parsed
+ with the current version of the Python client.
+ """
+
+ # deserialize JSON string
+ parsed_json = json.loads(json_string)
+
+ # check if the provided JSON has only one data series entry
+ # otherwise raise the proper exception
+ if len(parsed_json["allDataSeries"]) != 1:
+ raise StreamPipesUnsupportedDataLakeSeries()
+
+ # get the data data series
+ data_series = parsed_json["allDataSeries"][0]
+
+ return cls.parse_obj(data_series)
+
+ 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]
+ rows: List[List[Any]]
+ tags: Optional[str]
+
+ def to_pandas(self) -> pd.DataFrame:
+ """Returns the data lake series in representation of a Pandas Dataframe.
+
+ Returns
+ -------
+ pd.DataFrame
+ """
+
+ pandas_representation = self.convert_to_pandas_representation()
+ return pd.DataFrame(data=pandas_representation["rows"], columns=pandas_representation["headers"])
diff --git a/streampipes-client-python/tests/client/test_client.py b/streampipes-client-python/tests/client/test_client.py
index ec7d30d82..0c4027647 100644
--- a/streampipes-client-python/tests/client/test_client.py
+++ b/streampipes-client-python/tests/client/test_client.py
@@ -29,6 +29,7 @@ class TestStreamPipesClient(TestCase):
client_config=StreamPipesClientConfig(
credential_provider=StreamPipesApiKeyCredentials(username="user", api_key="key"),
host_address="localhost",
+ https_disabled=True,
)
)
@@ -43,15 +44,15 @@ class TestStreamPipesClient(TestCase):
dictionary=result_headers,
)
self.assertTrue(isinstance(result.dataLakeMeasureApi, DataLakeMeasureEndpoint))
- self.assertEqual(result.base_api_path, "https://localhost:443/streampipes-backend/")
+ self.assertEqual(result.base_api_path, "http://localhost:80/streampipes-backend/")
def test_client_create(self):
result = StreamPipesClient.create(
client_config=StreamPipesClientConfig(
credential_provider=StreamPipesApiKeyCredentials(username="user", api_key="key"),
host_address="localhost",
- https_disabled=True,
- port=500,
+ https_disabled=False,
+ port=443,
)
)
@@ -66,4 +67,4 @@ class TestStreamPipesClient(TestCase):
dictionary=result_headers,
)
self.assertTrue(isinstance(result.dataLakeMeasureApi, DataLakeMeasureEndpoint))
- self.assertEqual(result.base_api_path, "http://localhost:500/streampipes-backend/")
+ self.assertEqual(result.base_api_path, "https://localhost:443/streampipes-backend/")
diff --git a/streampipes-client-python/tests/client/test_data_lake_series.py b/streampipes-client-python/tests/client/test_data_lake_series.py
new file mode 100644
index 000000000..862b4a1ff
--- /dev/null
+++ b/streampipes-client-python/tests/client/test_data_lake_series.py
@@ -0,0 +1,154 @@
+#
+# 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.
+#
+import json
+from unittest import TestCase
+from unittest.mock import MagicMock, call, patch
+
+from streampipes_client.client import StreamPipesClient
+from streampipes_client.client.client_config import StreamPipesClientConfig
+from streampipes_client.client.credential_provider import StreamPipesApiKeyCredentials
+from streampipes_client.model.resource.data_lake_series import (
+ StreamPipesUnsupportedDataLakeSeries,
+)
+
+
+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": [
+ "time",
+ "changeDetectedHigh",
+ "changeDetectedLow",
+ "cumSumHigh",
+ "cumSumLow",
+ "level",
+ "overflow",
+ "sensorId",
+ "underflow",
+ ],
+ "allDataSeries": [],
+ }
+
+ @patch("streampipes_client.client.client.Session", autospec=True)
+ def test_to_pandas(self, http_session: MagicMock):
+ http_session_mock = MagicMock()
+ http_session_mock.get.return_value.text = json.dumps(self.series_regular)
+ http_session.return_value = http_session_mock
+
+ client = StreamPipesClient(
+ client_config=StreamPipesClientConfig(
+ credential_provider=StreamPipesApiKeyCredentials(username="user", api_key="key"),
+ host_address="localhost",
+ )
+ )
+ result = client.dataLakeMeasureApi.get(identifier="test")
+
+ http_session.assert_has_calls(
+ [call().get(url="https://localhost:80/streampipes-backend/api/v4/datalake/measurements/test")],
+ any_order=True,
+ )
+
+ result_pd = result.to_pandas()
+
+ self.assertEqual(2, len(result_pd))
+ self.assertListEqual(
+ [
+ "time",
+ "changeDetectedHigh",
+ "changeDetectedLow",
+ "cumSumHigh",
+ "cumSumLow",
+ "level",
+ "overflow",
+ "sensorId",
+ "underflow",
+ ],
+ list(result_pd.columns),
+ )
+ self.assertEqual(73.37740325927734, result_pd["level"][0])
+
+ @patch("streampipes_client.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
+
+ client = StreamPipesClient(
+ client_config=StreamPipesClientConfig(
+ credential_provider=StreamPipesApiKeyCredentials(username="user", api_key="key"),
+ host_address="localhost",
+ )
+ )
+ with self.assertRaises(StreamPipesUnsupportedDataLakeSeries):
+ client.dataLakeMeasureApi.get(identifier="test")
[streampipes] 04/04: Merge remote-tracking branch 'bossenti/STREAMPIPES-607' into STREAMPIPES-607
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch STREAMPIPES-607
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit bd47fda94eece008ac783155d3896c49383fc5a6
Merge: cc4f870a5 fe6ff0f2a
Author: bossenti <bo...@posteo.de>
AuthorDate: Thu Nov 24 20:00:41 2022 +0100
Merge remote-tracking branch 'bossenti/STREAMPIPES-607' into STREAMPIPES-607
# Conflicts:
# streampipes-client-python/.pre-commit-config.yaml
# streampipes-client-python/README.md
# streampipes-client-python/setup.py
# streampipes-client-python/streampipes_client/client/client_config.py
# streampipes-client-python/streampipes_client/endpoint/data_lake_measure.py
# streampipes-client-python/streampipes_client/endpoint/endpoint.py
# streampipes-client-python/streampipes_client/model/common.py
# streampipes-client-python/streampipes_client/model/container/resource_container.py
# streampipes-client-python/tests/client/test_client.py
.github/workflows/pr-validation.yml | 71 ++++++++
DISCLAIMER | 5 -
LICENSE | 10 +-
NOTICE | 4 +-
README.md | 2 +-
RELEASE_VALIDATION.md | 4 +-
installer/README.md | 2 +-
.../cli/deploy/standalone/rocketmq/Dockerfile | 76 ++++++++
.../standalone/rocketmq/docker-compose.dev.yml | 33 ++++
.../deploy/standalone/rocketmq/docker-compose.yml | 82 +++++++++
pom.xml | 31 +++-
streampipes-client-python/.pre-commit-config.yaml | 8 +
streampipes-client-python/README.md | 4 +-
.../docs/getting-started/installation.md | 41 +++++
.../getting-started/quickstart.md} | 38 +---
.../{README.md => docs/index.md} | 8 +-
streampipes-client-python/docs/overrides/main.html | 47 +++++
.../docs/overrides/partials/footer.html | 33 +++-
.../docs/scripts/gen_ref_pages.py | 50 +++++
.../docs/stylesheets/extra.css | 21 +++
streampipes-client-python/mkdocs.yml | 81 +++++++++
streampipes-client-python/setup.py | 7 +-
.../streampipes_client/client/client_config.py | 3 +-
.../endpoint/data_lake_measure.py | 18 +-
.../streampipes_client/endpoint/endpoint.py | 2 +-
.../streampipes_client/model/common.py | 6 +
.../model/container/resource_container.py | 16 +-
.../model/resource/data_lake_series.py | 118 ++++++++++++
.../tests/client/test_client.py | 9 +-
.../tests/client/test_data_lake_series.py | 154 ++++++++++++++++
.../dataexplorer/commons/influx/InfluxStore.java | 5 +-
.../dataexplorer/sdk/DataLakeQueryBuilder.java | 6 +
streampipes-extensions/README.md | 2 +-
.../streampipes-connect-adapters-iiot/pom.xml | 4 +
.../connect/iiot/ConnectAdapterIiotInit.java | 2 +
.../protocol/stream/pulsar/PulsarConsumer.java | 85 ---------
.../protocol/stream/pulsar/PulsarProtocol.java | 67 ++++---
.../protocol/stream/rocketmq/RocketMQConsumer.java | 64 +++++++
.../protocol/stream/rocketmq/RocketMQProtocol.java | 174 ++++++++++++++++++
.../protocol/stream/rocketmq/RocketMQUtils.java | 47 +++++
.../documentation.md | 16 +-
.../icon.png | Bin 0 -> 12775 bytes
.../strings.en | 11 ++
.../streampipes-sinks-brokers-jvm/pom.xml | 4 +
.../sinks/brokers/jvm/BrokersJvmInit.java | 2 +
.../brokers/jvm/rocketmq/RocketMQParameters.java | 44 +++++
.../jvm/rocketmq/RocketMQPublisherSink.java | 123 +++++++++++++
.../documentation.md | 37 +++-
.../icon.png | Bin 0 -> 12775 bytes
.../strings.en | 8 +
.../jvm/rocketmq/TestRocketMQPublisherSink.java | 83 +++++++++
.../integration/adapters/PulsarAdapterTester.java | 2 +
.../streampipes/rest/impl/PipelineResource.java | 26 +++
streampipes-wrapper-python/LICENSE | 201 ---------------------
streampipes-wrapper-python/NOTICE | 7 -
streampipes-wrapper-python/README.md | 12 +-
ui/package.json | 2 +-
57 files changed, 1597 insertions(+), 421 deletions(-)
diff --cc streampipes-client-python/streampipes_client/client/client_config.py
index ee98e6bf9,0e2a38144..96ff045a7
--- a/streampipes-client-python/streampipes_client/client/client_config.py
+++ b/streampipes-client-python/streampipes_client/client/client_config.py
@@@ -45,7 -46,7 +46,7 @@@ class StreamPipesClientConfig
https_disabled: Optional[bool]
Determines whether https is used to connect to StreamPipes.
port: Optional[int]
- Specifies the port under which the StreamPipes API is available, e.g., 80.
- Specifies the port under which the StreamPipes API is available, e.g., 80 (with http) or 443 (with https)
++ Specifies the port under which the StreamPipes API is available, e.g., `80` (with http) or `443` (with https)
Examples
--------