(streampipes) branch dev updated: fix(#2551): prevent Python functions from creating multiple output streams (#2561)

[bo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git


The following commit(s) were added to refs/heads/dev by this push:
     new 3e17c932a6 fix(#2551): prevent Python functions from creating multiple output streams  (#2561)
3e17c932a6 is described below

commit 3e17c932a666e7d82f58aa5cf4ea3a29fe71ccf2
Author: Tim <50...@users.noreply.github.com>
AuthorDate: Fri Mar 15 09:59:48 2024 +0100

    fix(#2551): prevent Python functions from creating multiple output streams  (#2561)
    
    * fix(#2551): create reliable data streams
    
    * fix(#2551): provide default value for element_id
    
    * fix(#2551): improve exception handling
    
    * fix(#2551): change log level to info
---
 .../streampipes/endpoint/endpoint.py                   |  4 +++-
 .../streampipes/functions/function_handler.py          | 18 +++++++++++++++---
 .../streampipes/model/resource/data_stream.py          |  5 +++--
 3 files changed, 21 insertions(+), 6 deletions(-)

diff --git a/streampipes-client-python/streampipes/endpoint/endpoint.py b/streampipes-client-python/streampipes/endpoint/endpoint.py
index b3572d9d11..fb70fbf41a 100644
--- a/streampipes-client-python/streampipes/endpoint/endpoint.py
+++ b/streampipes-client-python/streampipes/endpoint/endpoint.py
@@ -44,6 +44,7 @@ logger = logging.getLogger(__name__)
 
 # define custom logging messages for some specific HTTP status
 _error_code_to_message = {
+    400: "\nThe StreamPipes Backend did not accept the request. Please check the logs for details.\n",
     401: "\nThe StreamPipes Backend returned an unauthorized error.\n"
     "Please check your user name and/or password to be correct.",
     403: "\nThere seems to be an issue with the access rights of the given user and the resource you queried.\n"
@@ -149,13 +150,14 @@ class APIEndpoint(Endpoint):
             error_message = _error_code_to_message[status_code]
 
             if status_code in [
+                HTTPStatus.BAD_REQUEST.numerator,
                 HTTPStatus.METHOD_NOT_ALLOWED.numerator,
                 HTTPStatus.NOT_FOUND.numerator,
             ]:
                 error_message += f"url: {err.response.url}\nstatus code: {status_code}"
 
             logger.debug(f"HTTP error response: {err.response.text}")
-            raise HTTPError(error_message) from err
+            raise HTTPError(error_message, response=response) from err
 
         else:
             logger.debug("Successfully retrieved resources from %s.", url)
diff --git a/streampipes-client-python/streampipes/functions/function_handler.py b/streampipes-client-python/streampipes/functions/function_handler.py
index 4eaceb1754..1a60bbb336 100644
--- a/streampipes-client-python/streampipes/functions/function_handler.py
+++ b/streampipes-client-python/streampipes/functions/function_handler.py
@@ -17,8 +17,11 @@
 import asyncio
 import json
 import logging
+from http import HTTPStatus
 from typing import AsyncIterator, Dict, List
 
+from requests import HTTPError
+
 from streampipes.client.client import StreamPipesClient
 from streampipes.functions.broker import Broker, Consumer, get_broker
 from streampipes.functions.registration import Registration
@@ -67,10 +70,19 @@ class FunctionHandler:
         for streampipes_function in self.registration.getFunctions():
             # Create the output data streams for every function
             for stream_id, output_stream in streampipes_function.function_definition.get_output_data_streams().items():
-                self.client.dataStreamApi.post(output_stream)
+                try:
+                    self.client.dataStreamApi.post(output_stream)
+                except HTTPError as e:
+                    logger.info("The data stream could not be created.")
+                    if e.response.status_code == HTTPStatus.BAD_REQUEST:
+                        logger.info(
+                            "This is due to the fact that this data stream already exists. "
+                            "Continuing with the existing data stream."
+                        )
+                    else:
+                        raise RuntimeError from e
                 logger.info(
-                    f'Create output data stream "{stream_id}" '
-                    f'for the function "{streampipes_function.getFunctionId().id}"'
+                    f"Using output data stream '{stream_id}' for function '{streampipes_function.getFunctionId().id}'"
                 )
             # Choose the broker and collect the schema for every data stream
             for stream_id in streampipes_function.requiredStreamIds():
diff --git a/streampipes-client-python/streampipes/model/resource/data_stream.py b/streampipes-client-python/streampipes/model/resource/data_stream.py
index 10ea2b34d6..d8e424c4b0 100644
--- a/streampipes-client-python/streampipes/model/resource/data_stream.py
+++ b/streampipes-client-python/streampipes/model/resource/data_stream.py
@@ -24,7 +24,6 @@ from streampipes.model.common import (
     EventSchema,
     MeasurementCapability,
     MeasurementObject,
-    random_letters,
 )
 from streampipes.model.resource.resource import Resource
 
@@ -81,11 +80,13 @@ class DataStream(Resource):
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
+        if not self.element_id:
+            self.element_id = f"sp:spdatastream:{self.name.replace(':', '')}"
         if not self.uri:
             self.uri = self.element_id
 
     class_name: StrictStr = Field(alias="@class", default_factory=lambda: "org.apache.streampipes.model.SpDataStream")
-    element_id: StrictStr = Field(default_factory=lambda: f"sp:spdatastream:{random_letters(6)}")
+    element_id: StrictStr = Field(default="")
     name: StrictStr = Field(default="Unnamed")
     description: Optional[StrictStr]
     icon_url: Optional[StrictStr]