Bugfix/allow removing outputs from services (#2063)

* correctly handle removal of files from outputs
* ensure frontend removes deleted outputs

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/api/routes/computations.py

@@ -34,7 +34,7 @@
     is_pipeline_stopped,
 )
 from ...utils.dags import create_dag_graph, create_minimal_graph_based_on_selection
-from ...utils.exceptions import ProjectNotFoundError
+from ...utils.exceptions import PipelineNotFoundError, ProjectNotFoundError
 from ..dependencies.celery import CeleryClient, get_celery_client
 from ..dependencies.database import get_repository
 from ..dependencies.director_v0 import DirectorV0Client, get_director_v0_client
@@ -228,7 +228,7 @@ async def get_computation(
         )
         return task_out
 
-    except ProjectNotFoundError as e:
+    except (ProjectNotFoundError, PipelineNotFoundError) as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e
 
     # NOTE: this will be re-used for the prep2go API stuff... don't worry...

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_pipelines.py

@@ -8,6 +8,7 @@
 from sqlalchemy.dialects.postgresql import insert
 
 from ....models.domains.comp_pipelines import CompPipelineAtDB
+from ....utils.exceptions import PipelineNotFoundError
 from ....utils.logging_utils import log_decorator
 from ..tables import comp_pipeline
 from ._base import BaseRepository
@@ -24,6 +25,8 @@ async def get_pipeline(self, project_id: ProjectID) -> CompPipelineAtDB:
             )
         )
         row: RowProxy = await result.fetchone()
+        if not row:
+            raise PipelineNotFoundError(str(project_id))
         return CompPipelineAtDB.from_orm(row)
 
     @log_decorator(logger=logger)

services/director-v2/src/simcore_service_director_v2/utils/exceptions.py

@@ -11,7 +11,14 @@ def __init__(self, msg: Optional[str] = None):
 
 
 class ProjectNotFoundError(DirectorException):
-    """Service was not found in swarm"""
+    """Project not found error"""
 
     def __init__(self, project_id: ProjectID):
         super().__init__(f"project {project_id} not found")
+
+
+class PipelineNotFoundError(DirectorException):
+    """Pipeline not found error"""
+
+    def __init__(self, pipeline_id: str):
+        super().__init__(f"pipeline {pipeline_id} not found")

services/web/client/source/class/osparc/data/model/Node.js

@@ -610,11 +610,15 @@ qx.Class.define("osparc.data.model.Node", {
 
     setOutputData: function(outputs) {
       if (outputs) {
-        for (const outputKey in outputs) {
+        for (const outputKey in this.__outputs) {
           if (!Object.prototype.hasOwnProperty.call(this.__outputs, outputKey)) {
             this.__outputs[outputKey] = {};
           }
-          this.__outputs[outputKey]["value"] = outputs[outputKey];
+          if (Object.prototype.hasOwnProperty.call(outputs, outputKey)) {
+            this.__outputs[outputKey]["value"] = outputs[outputKey];
+          } else {
+            this.__outputs[outputKey]["value"] = "";
+          }
           this.fireDataEvent("outputChanged", outputKey);
         }
       }

services/web/server/src/simcore_service_webserver/computation_comp_tasks_listening_task.py

@@ -6,7 +6,7 @@
 import json
 import logging
 from pprint import pformat
-from typing import Dict, List
+from typing import Dict
 
 from aiohttp import web
 from aiopg.sa import Engine
@@ -61,11 +61,8 @@ async def _update_project_outputs(
     node_uuid: NodeID,
     outputs: Dict,
 ) -> None:
-    changed_keys: List[str] = list(outputs.keys())
-    if not changed_keys:
-        return
-
-    project = await projects_api.update_project_node_outputs(
+    # the new outputs might be {}, or {key_name: payload}
+    project, changed_keys = await projects_api.update_project_node_outputs(
         app,
         user_id,
         project_uuid,

services/web/server/src/simcore_service_webserver/projects/projects_api.py

@@ -11,7 +11,7 @@
 import logging
 from collections import defaultdict
 from pprint import pformat
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional, Set, Tuple
 from uuid import uuid4
 
 from aiohttp import web
@@ -344,7 +344,7 @@ async def update_project_node_outputs(
     project_id: str,
     node_id: str,
     data: Optional[Dict],
-) -> Dict:
+) -> Tuple[Dict, List[str]]:
     """
     Updates outputs of a given node in a project with 'data'
     """
@@ -363,25 +363,34 @@ async def update_project_node_outputs(
 
     # NOTE: update outputs (not required) if necessary as the UI expects a
     # dataset/label field that is missing
-    outputs: Dict[str, Any] = project["workbench"][node_id].setdefault("outputs", {})
-    outputs.update(data)
-
-    for output_key in outputs.keys():
-        if not isinstance(outputs[output_key], dict):
+    current_outputs = project["workbench"][node_id].setdefault("outputs", {})
+    new_outputs = data
+    project["workbench"][node_id]["outputs"] = new_outputs
+
+    # find changed keys (the ones that appear or disapppear for sure)
+    changed_keys = list(current_outputs.keys() ^ new_outputs.keys())
+    # now check the ones that are in both object
+    for key in current_outputs.keys() & new_outputs.keys():
+        if current_outputs[key] != new_outputs[key]:
+            changed_keys.append(key)
+
+    # FIXME: this should be reviewed @maiz. how is an output file defined. I think we have several flavours.
+    for output_key in new_outputs.keys():
+        if not isinstance(new_outputs[output_key], dict):
             continue
-        if "path" in outputs[output_key]:
+        if "path" in new_outputs[output_key]:
             # file_id is of type study_id/node_id/file.ext
-            file_id = outputs[output_key]["path"]
+            file_id = new_outputs[output_key]["path"]
             study_id, _, file_ext = file_id.split("/")
-            outputs[output_key]["dataset"] = study_id
-            outputs[output_key]["label"] = file_ext
+            new_outputs[output_key]["dataset"] = study_id
+            new_outputs[output_key]["label"] = file_ext
 
     db = app[APP_PROJECT_DBAPI]
     updated_project = await db.update_user_project(project, user_id, project_id)
     updated_project["state"] = await get_project_state_for_user(
         user_id, project_id, app
     )
-    return updated_project
+    return updated_project, changed_keys
 
 
 async def get_workbench_node_ids_from_project_uuid(