SNRGY-2824 validate operators with operator-repo

main.py

@@ -23,6 +23,8 @@
 import jwt
 from pymongo import MongoClient, ReturnDocument, ASCENDING, DESCENDING
 
+from operators import get_operator
+
 app = Flask("analytics-flow-repo")
 app.config.SWAGGER_UI_DOC_EXPANSION = 'list'
 CORS(app)
@@ -93,6 +95,9 @@ def put(self):
         """Creates a flow."""
         user_id = get_user_id(request)
         req = request.get_json()
+        code = fill_operator_info(req, user_id)
+        if code != 200:
+            return None, code
         req['userId'] = user_id
         req['dateCreated'] = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).isoformat()
         req['dateUpdated'] = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).isoformat()
@@ -163,6 +168,9 @@ def post(self, flow_id):
         """Updates a flow."""
         user_id = get_user_id(request)
         req = request.get_json()
+        code = fill_operator_info(req, user_id)
+        if code != 200:
+            return "error", code
         req['dateUpdated'] = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).isoformat()
         flow = flows.find_one_and_update({'$and': [{'_id': ObjectId(flow_id)},
                                                    {'$or': [{'userId': user_id}, {'share.write': True}]}]}, {
@@ -195,6 +203,29 @@ def get_user_id(req):
     return user_id
 
 
+def fill_operator_info(flow, user_id) -> int :
+    if "model" not in flow:
+        return 400
+    model = flow["model"]
+    if "cells" not in model:
+        return 400
+    cells = model["cells"]
+    for cell in cells:
+        if "type" not in cell:
+            return 400
+        if cell["type"] != "senergy.NodeElement":
+            continue
+        if "operatorId" not in cell:
+            return 400
+        operator, code = get_operator(cell["operatorId"], user_id)
+        if code != 200:
+            return code
+        cell["name"] = operator["name"]
+        cell["image"] = operator["image"]
+        cell["deploymentType"] = operator["deploymentType"]
+        cell["cost"] = operator["cost"]
+    return 200
+
 if bool(os.getenv('DEBUG', '')):
     if __name__ == "__main__":
         app.run("127.0.0.1", 5000, debug=True)

operators.py

@@ -0,0 +1,49 @@
+#  Copyright 2018 InfAI (CC SES)
+#
+#  Licensed 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 os
+import typing
+
+import requests
+
+url = os.getenv('OPERATOR_REPO_URL', 'localhost')
+
+
+class OperatorIO(typing.TypedDict):
+    name: str
+    type: str
+
+
+class Operator(typing.TypedDict, total=False):
+    name: str
+    image: typing.NotRequired[str]
+    description: typing.NotRequired[str]
+    pub: typing.NotRequired[bool]
+    deploymentType: typing.NotRequired[str]
+    inputs: typing.List[OperatorIO]
+    outputs: typing.List[OperatorIO]
+    cost: typing.NotRequired[int]
+    config_values: typing.List[OperatorIO]
+
+
+def get_operator(operator_id: str, user_id: str) -> typing.Tuple[typing.Optional[Operator], int]:
+    try:
+        resp = requests.get(url + '/operator/' + operator_id, timeout=10, headers={'X-UserID' : user_id})
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching operator {operator_id} for user {user_id} with exception {e}")
+        return None, 502
+    if resp.status_code != 200:
+        return None, resp.status_code
+    return resp.json(), 200