chore: [VRD-1075] Update orchestrator for new pipeline defn (#4013)

Author: liuverta

client/verta/tests/pipeline/test_orchestrator.py

@@ -72,21 +72,20 @@ def test_run_and_logs(self, registered_model):
         triple_model_ver = create_standard_model(Triple)
         sum_model_ver = create_standard_model(Sum)
 
-        pipeline_spec = {
-            "pipeline_name": "diamond-pipeline",
+        pipeline_defn = {
             "steps": [
                 {"name": "echo", "model_version_id": echo_model_ver.id},
                 {"name": "double", "model_version_id": double_model_ver.id},
                 {"name": "triple", "model_version_id": triple_model_ver.id},
                 {"name": "sum", "model_version_id": sum_model_ver.id},
             ],
             "graph": [
-                {"name": "double", "inputs": ["echo"]},
-                {"name": "triple", "inputs": ["echo"]},
-                {"name": "sum", "inputs": ["double", "triple"]},
+                {"name": "double", "predecessors": ["echo"]},
+                {"name": "triple", "predecessors": ["echo"]},
+                {"name": "sum", "predecessors": ["double", "triple"]},
             ],
         }
-        orchestrator = LocalOrchestrator(registered_model._conn, pipeline_spec)
+        orchestrator = LocalOrchestrator(registered_model._conn, pipeline_defn)
 
         input = 3
         with runtime.context() as ctx:

client/verta/verta/__about__.py

@@ -6,4 +6,4 @@
 __maintainer_email__ = "[email protected]"
 __title__ = "verta"
 __url__ = "https://www.verta.ai/"
-__version__ = "0.24.0"
+__version__ = "0.24.1a0"

client/verta/verta/_pipeline_orchestrator/_orchestrator.py

@@ -17,10 +17,10 @@ class _OrchestratorBase(abc.ABC):
 
     def __init__(
         self,
-        pipeline_spec: Dict[str, Any],
+        pipeline_defn: Dict[str, Any],
         step_handlers: Dict[str, _StepHandlerBase],
     ):
-        self._pipeline_spec = pipeline_spec
+        self._pipeline_defn = pipeline_defn
         self._step_handlers = step_handlers
 
         # DAG nodes are step names
@@ -30,13 +30,13 @@ def __init__(
 
     @staticmethod
     def _get_predecessors_mapping(
-        pipeline_spec: Dict[str, Any],
+        pipeline_defn: Dict[str, Any],
     ) -> Dict[str, List[str]]:
         """Get the names of steps' predecessors from a pipeline specification.
 
         Parameters
         ----------
-        pipeline_spec : dict
+        pipeline_defn : dict
             Pipeline specification.
 
         Returns
@@ -45,14 +45,14 @@ def _get_predecessors_mapping(
             Mapping from step names to their predecessors' names.
 
         """
-        return {node["name"]: node["inputs"] for node in pipeline_spec["graph"]}
+        return {node["name"]: node["predecessors"] for node in pipeline_defn["graph"]}
 
     def _prepare_pipeline(self):
         """Initialize ``self._dag`` and ``self._outputs``.
 
         Parameters
         ----------
-        pipeline_spec : dict
+        pipeline_defn : dict
             Pipeline specification.
 
         Raises
@@ -61,7 +61,7 @@ def _prepare_pipeline(self):
             If the pipeline graph has cycles.
 
         """
-        dag = TopologicalSorter(self._get_predecessors_mapping(self._pipeline_spec))
+        dag = TopologicalSorter(self._get_predecessors_mapping(self._pipeline_defn))
         dag.prepare()
         # TODO: assert one input node
         # TODO: assert one output node
@@ -201,7 +201,7 @@ class LocalOrchestrator(_OrchestratorBase):
     ----------
     conn : :class:`~verta._internal_utils._utils.Connection`
         Verta client connection.
-    pipeline_spec : dict
+    pipeline_defn : dict
         Pipeline specification.
 
     Examples
@@ -210,34 +210,34 @@ class LocalOrchestrator(_OrchestratorBase):
 
         from verta._pipeline_orchestrator import LocalOrchestrator
 
-        orchestrator = LocalOrchestrator(client._conn, pipeline_spec)
+        orchestrator = LocalOrchestrator(client._conn, pipeline_defn)
         pipeline_output = orchestrator.run(pipeline_input)
 
     """
 
     def __init__(
         self,
         conn: _utils.Connection,
-        pipeline_spec: Dict[str, Any],
+        pipeline_defn: Dict[str, Any],
     ):
         super().__init__(
-            pipeline_spec=pipeline_spec,
-            step_handlers=self._init_step_handlers(conn, pipeline_spec),
+            pipeline_defn=pipeline_defn,
+            step_handlers=self._init_step_handlers(conn, pipeline_defn),
         )
 
     @classmethod
     def _init_step_handlers(
         cls,
         conn: _utils.Connection,
-        pipeline_spec: Dict[str, Any],
+        pipeline_defn: Dict[str, Any],
     ) -> Dict[str, ModelObjectStepHandler]:
         """Instantiate and return step handlers.
 
         Parameters
         ----------
         conn : :class:`~verta._internal_utils._utils.Connection`
             Verta client connection.
-        pipeline_spec : dict
+        pipeline_defn : dict
             Pipeline specification.
 
         Returns
@@ -246,10 +246,10 @@ def _init_step_handlers(
             Mapping of step names to their handlers.
 
         """
-        predecessors_mapping = cls._get_predecessors_mapping(pipeline_spec)
+        predecessors_mapping = cls._get_predecessors_mapping(pipeline_defn)
 
         step_handlers = dict()
-        for step in pipeline_spec["steps"]:
+        for step in pipeline_defn["steps"]:
             step_handlers[step["name"]] = ModelObjectStepHandler(
                 name=step["name"],
                 predecessors=predecessors_mapping.get(step["name"], []),