mode PipelineInfo model into models module @sanderegg

Author: bisgaard-itis

services/director-v2/src/simcore_service_director_v2/models/comp_tasks.py

@@ -2,6 +2,7 @@
 from contextlib import suppress
 from typing import Annotated, Any
 
+import networkx as nx
 from dask_task_models_library.container_tasks.protocol import ContainerEnvsDict
 from models_library.api_schemas_directorv2.services import NodeRequirements
 from models_library.basic_regex import SIMPLE_VERSION_RE
@@ -276,3 +277,9 @@ class ComputationTaskForRpcDBGet(BaseModel):
     image: dict[str, Any]
     started_at: dt.datetime | None
     ended_at: dt.datetime | None
+
+
+class PipelineInfo(BaseModel):
+    pipeline_dag: nx.DiGraph
+    all_tasks: list[CompTaskAtDB]
+    filtered_tasks: list[CompTaskAtDB]

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

@@ -1,22 +1,13 @@
-from typing import NamedTuple
-
 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
+from ..models.comp_tasks import CompTaskAtDB, PipelineInfo
 from ..modules.db.repositories.comp_pipelines import CompPipelinesRepository
 from ..modules.db.repositories.comp_tasks import CompTasksRepository
 
 
-class PipelineInfo(NamedTuple):
-    # NOTE: kept old names for legacy but should rename for clarity
-    pipeline_dag: nx.DiGraph
-    all_tasks: list[CompTaskAtDB]  # all nodes in pipeline
-    filtered_tasks: list[CompTaskAtDB]  # nodes that actually run i.e. part of the dag
-
-
 async def _get_pipeline_info(
     *,
     project_id: ProjectID,
@@ -39,7 +30,9 @@ async def _get_pipeline_info(
         t for t in all_tasks if f"{t.node_id}" in set(pipeline_dag.nodes())
     ]
 
-    return PipelineInfo(pipeline_dag, all_tasks, filtered_tasks)
+    return PipelineInfo(
+        pipeline_dag=pipeline_dag, all_tasks=all_tasks, filtered_tasks=filtered_tasks
+    )
 
 
 async def validate_pipeline(