also validate pipeline in rpc case @sanderegg

Author: bisgaard-itis

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

@@ -24,7 +24,7 @@
 from ...modules.db.repositories.comp_pipelines import CompPipelinesRepository
 from ...modules.db.repositories.comp_tasks import CompTasksRepository
 from ...utils import dask as dask_utils
-from ...utils.computations_tasks import PipelineInfo, get_pipeline_info
+from ...utils.computations_tasks import PipelineTaskMissingError, validate_pipeline
 from ..dependencies.database import get_repository
 
 log = logging.getLogger(__name__)
@@ -35,31 +35,6 @@
 # HELPERS -------------------------------------------------------------------
 
 
-async def validate_pipeline(
-    project_id: ProjectID,
-    comp_pipelines_repo: CompPipelinesRepository,
-    comp_tasks_repo: CompTasksRepository,
-) -> PipelineInfo:
-    """
-    Loads and validates data from pipelines and tasks tables and
-    reports it back as PipelineInfo
-    """
-
-    pipeline_info = await get_pipeline_info(
-        project_id=project_id,
-        comp_pipelines_repo=comp_pipelines_repo,
-        comp_tasks_repo=comp_tasks_repo,
-    )
-
-    # check that we have the expected tasks
-    if len(pipeline_info.filtered_tasks) != len(pipeline_info.pipeline_dag):
-        raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
-            detail="The tasks referenced by the pipeline are missing",
-        )
-    return pipeline_info
-
-
 # ROUTES HANDLERS --------------------------------------------------------------
 
 
@@ -82,7 +57,13 @@ async def get_all_tasks_log_files(
     Each log is only available when the corresponding task is done
     """
     # gets computation task ids
-    info = await validate_pipeline(project_id, comp_pipelines_repo, comp_tasks_repo)
+    try:
+        info = await validate_pipeline(project_id, comp_pipelines_repo, comp_tasks_repo)
+    except PipelineTaskMissingError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="The tasks referenced by the pipeline are missing",
+        ) from exc
     iter_task_ids = (t.node_id for t in info.filtered_tasks)
 
     tasks_logs_files: list[TaskLogFileGet] = await logged_gather(

services/director-v2/src/simcore_service_director_v2/api/rpc/_computations_tasks.py

@@ -10,10 +10,10 @@
 from simcore_sdk.node_ports_common import data_items_utils
 
 from ...constants import LOGS_FILE_NAME
-from ...core.errors import PipelineNotFoundError
+from ...core.errors import PipelineNotFoundError, PipelineTaskMissingError
 from ...modules.db.repositories.comp_pipelines import CompPipelinesRepository
 from ...modules.db.repositories.comp_tasks import CompTasksRepository
-from ...utils.computations_tasks import get_pipeline_info
+from ...utils.computations_tasks import validate_pipeline
 
 router = RPCRouter()
 
@@ -27,12 +27,12 @@ async def get_computation_task_log_file_ids(
     comp_tasks_repo = CompTasksRepository.instance(db_engine=app.state.engine)
 
     try:
-        info = await get_pipeline_info(
+        info = await validate_pipeline(
             project_id=project_id,
             comp_pipelines_repo=comp_pipelines_repo,
             comp_tasks_repo=comp_tasks_repo,
         )
-    except PipelineNotFoundError as exc:
+    except (PipelineNotFoundError, PipelineTaskMissingError) as exc:
         raise ComputationalTaskMissingError(project_id=project_id) from exc
 
     iter_task_ids = (t.node_id for t in info.filtered_tasks)

services/director-v2/src/simcore_service_director_v2/core/errors.py

@@ -67,6 +67,10 @@ class WalletNotEnoughCreditsError(DirectorError):
     msg_template = "Wallet '{wallet_name}' has {wallet_credit_amount} credits."
 
 
+class PipelineTaskMissingError(DirectorError):
+    msg_template = "Pipeline associated with project_id {project_id} is missing task(s)"
+
+
 #
 # SCHEDULER ERRORS
 #

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

@@ -2,6 +2,7 @@
 
 import networkx as nx
 from models_library.projects import ProjectID
+from simcore_service_director_v2.core.errors import PipelineTaskMissingError
 
 from ..models.comp_pipelines import CompPipelineAtDB
 from ..models.comp_tasks import CompTaskAtDB
@@ -16,7 +17,7 @@ class PipelineInfo(NamedTuple):
     filtered_tasks: list[CompTaskAtDB]  # nodes that actually run i.e. part of the dag
 
 
-async def get_pipeline_info(
+async def _get_pipeline_info(
     *,
     project_id: ProjectID,
     comp_pipelines_repo: CompPipelinesRepository,
@@ -39,3 +40,28 @@ async def get_pipeline_info(
     ]
 
     return PipelineInfo(pipeline_dag, all_tasks, filtered_tasks)
+
+
+async def validate_pipeline(
+    project_id: ProjectID,
+    comp_pipelines_repo: CompPipelinesRepository,
+    comp_tasks_repo: CompTasksRepository,
+) -> PipelineInfo:
+    """
+    Loads and validates data from pipelines and tasks tables and
+    reports it back as PipelineInfo
+
+    raises PipelineTaskMissingError
+    """
+
+    pipeline_info = await _get_pipeline_info(
+        project_id=project_id,
+        comp_pipelines_repo=comp_pipelines_repo,
+        comp_tasks_repo=comp_tasks_repo,
+    )
+
+    # check that we have the expected tasks
+    if len(pipeline_info.filtered_tasks) != len(pipeline_info.pipeline_dag):
+        raise PipelineTaskMissingError(project_id=project_id)
+
+    return pipeline_info