Refactoring

Author: schustmi

src/zenml/config/compiler.py

@@ -489,15 +489,15 @@ def _compile_step_invocation(
         invocation.step = copy.deepcopy(invocation.step)
 
         step = invocation.step
-        with step._skip_dynamic_configuration():
+        with step._suspend_dynamic_configuration():
             if step_config:
                 step._apply_configuration(
                     step_config, runtime_parameters=invocation.parameters
                 )
 
             # Apply the dynamic configuration (which happened while executing the
             # pipeline function) after all other step-specific configurations.
-            step._apply_dynamic_configuration()
+            step._merge_dynamic_configuration()
 
             convert_component_shortcut_settings_keys(
                 step.configuration.settings, stack=stack

src/zenml/orchestrators/base_orchestrator.py

@@ -213,7 +213,18 @@ def submit_dynamic_pipeline(
         environment: Dict[str, str],
         placeholder_run: Optional["PipelineRunResponse"] = None,
     ) -> Optional[SubmissionResult]:
-        """Submits a dynamic pipeline to the orchestrator."""
+        """Submits a dynamic pipeline to the orchestrator.
+
+        Args:
+            snapshot: The pipeline snapshot to submit.
+            stack: The stack the pipeline will run on.
+            environment: Environment variables to set in the orchestration
+                environment.
+            placeholder_run: An optional placeholder run.
+
+        Returns:
+            Optional submission result.
+        """
         return None
 
     def prepare_or_run_pipeline(
@@ -429,6 +440,8 @@ def run_step(
         """
         from zenml.pipelines.dynamic.runner import _run_step_sync
 
+        assert self._active_snapshot
+
         _run_step_sync(
             snapshot=self._active_snapshot,
             step=step,
@@ -438,23 +451,41 @@ def run_step(
 
     @property
     def supports_dynamic_pipelines(self) -> bool:
+        """Whether the orchestrator supports dynamic pipelines.
+
+        Returns:
+            Whether the orchestrator supports dynamic pipelines.
+        """
         return (
             getattr(self.submit_dynamic_pipeline, "__func__", None)
             is not BaseOrchestrator.submit_dynamic_pipeline
         )
 
     @property
-    def supports_dynamic_out_of_process_steps(self) -> bool:
+    def can_launch_dynamic_steps(self) -> bool:
+        """Whether the orchestrator can launch dynamic steps.
+
+        Returns:
+            Whether the orchestrator can launch dynamic steps.
+        """
         return (
-            getattr(self.run_dynamic_out_of_process_step, "__func__", None)
-            is not BaseOrchestrator.run_dynamic_out_of_process_step
+            getattr(self.launch_dynamic_step, "__func__", None)
+            is not BaseOrchestrator.launch_dynamic_step
         )
 
-    def run_dynamic_out_of_process_step(
+    def launch_dynamic_step(
         self, step_run_info: "StepRunInfo", environment: Dict[str, str]
     ) -> None:
+        """Launch a dynamic step.
+
+        Args:
+            step_run_info: The step run information.
+            environment: The environment variables to set in the execution
+                environment.
+        """
         raise NotImplementedError(
-            "Running dynamic out of process steps is not implemented for the orchestrator."
+            "Launching dynamic steps is not implemented for "
+            f"the {self.__class__.__name__} orchestrator."
         )
 
     @staticmethod

src/zenml/orchestrators/step_launcher.py

@@ -481,8 +481,7 @@ def _run_step(
                         # the orchestrator doesn't support it.
                         logger.warning(
                             "The %s does not support running dynamic out of "
-                            "process steps. Running step `%s` in current "
-                            "thread instead.",
+                            "process steps. Running step `%s` locally instead.",
                             self._stack.orchestrator.__class__.__name__,
                             self._invocation_id,
                         )
@@ -574,7 +573,7 @@ def _run_step_with_dynamic_orchestrator(
                 stack=self._stack,
             )
         )
-        self._stack.orchestrator.run_dynamic_out_of_process_step(
+        self._stack.orchestrator.launch_dynamic_step(
             step_run_info=step_run_info,
             environment=environment,
         )

src/zenml/pipelines/dynamic/pipeline_definition.py

@@ -19,54 +19,70 @@
     Dict,
     List,
     Optional,
-    Set,
     Type,
-    Union,
 )
 
-from pydantic import BaseModel, ConfigDict, ValidationError, create_model
+from pydantic import BaseModel, ConfigDict, create_model
 
-from zenml import ExternalArtifact
 from zenml.client import Client
 from zenml.logger import get_logger
-from zenml.models import ArtifactVersionResponse, PipelineRunResponse
+from zenml.models import PipelineRunResponse
 from zenml.pipelines.pipeline_definition import Pipeline
 from zenml.pipelines.run_utils import (
     should_prevent_pipeline_execution,
 )
-from zenml.steps.step_invocation import StepInvocation
 from zenml.steps.utils import (
     parse_return_type_annotations,
 )
-from zenml.utils import dict_utils, pydantic_utils
 
 if TYPE_CHECKING:
     from zenml.steps import BaseStep
-    from zenml.steps.entrypoint_function_utils import StepArtifact
 
 logger = get_logger(__name__)
 
 
 class DynamicPipeline(Pipeline):
     """Dynamic pipeline class."""
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        *args: Any,
+        depends_on: Optional[List["BaseStep"]] = None,
+        **kwargs: Any,
+    ) -> None:
         """Initialize the pipeline.
 
         Args:
             *args: Pipeline constructor arguments.
+            depends_on: The steps that the pipeline depends on.
             **kwargs: Pipeline constructor keyword arguments.
+        """
+        super().__init__(*args, **kwargs)
+        self._depends_on = depends_on or []
+        self._validate_depends_on(self._depends_on)
+
+    def _validate_depends_on(self, depends_on: List["BaseStep"]) -> None:
+        """Validates the steps that the pipeline depends on.
+
+        Args:
+            depends_on: The steps that the pipeline depends on.
 
         Raises:
-            ValueError: If some of the steps in `depends_on` are duplicated.
+            RuntimeError: If some of the steps in `depends_on` are duplicated.
         """
-        self._depends_on = kwargs.pop("depends_on", None) or []
-        if self._depends_on:
-            static_ids = [step._static_id for step in self._depends_on]
-            if len(static_ids) != len(set(static_ids)):
-                raise ValueError("Duplicate steps in depends_on.")
+        static_ids = set()
+        for step in depends_on:
+            static_id = step._static_id
+            if static_id in static_ids:
+                raise RuntimeError(
+                    f"The pipeline {self.name} depends on the same step "
+                    f"({step.name}) multiple times. To fix this, remove the "
+                    "duplicate from the `depends_on` list. You can pass the "
+                    "same step function with multiple configurations by using "
+                    "the `step.with_options(...)` method."
+                )
 
-        super().__init__(*args, **kwargs)
+            static_ids.add(static_id)
 
     @property
     def depends_on(self) -> List["BaseStep"]:
@@ -86,137 +102,19 @@ def is_dynamic(self) -> bool:
         """
         return True
 
-    @property
-    def is_prepared(self) -> bool:
-        """If the pipeline is prepared.
-
-        Prepared means that the pipeline entrypoint has been called and the
-        pipeline is fully defined.
-
-        Returns:
-            If the pipeline is prepared.
-        """
-        return False
-
-    def prepare(self, *args: Any, **kwargs: Any) -> None:
-        """Prepares the pipeline.
-
-        Args:
-            *args: Pipeline entrypoint input arguments.
-            **kwargs: Pipeline entrypoint input keyword arguments.
-
-        Raises:
-            RuntimeError: If the pipeline has parameters configured differently in
-                configuration file and code.
-        """
-        conflicting_parameters = {}
-        parameters_ = (self.configuration.parameters or {}).copy()
-        if from_file_ := self._from_config_file.get("parameters", None):
-            parameters_ = dict_utils.recursive_update(parameters_, from_file_)
-        if parameters_:
-            for k, v_runtime in kwargs.items():
-                if k in parameters_:
-                    v_config = parameters_[k]
-                    if v_config != v_runtime:
-                        conflicting_parameters[k] = (v_config, v_runtime)
-            if conflicting_parameters:
-                is_plural = "s" if len(conflicting_parameters) > 1 else ""
-                msg = f"Configured parameter{is_plural} for the pipeline `{self.name}` conflict{'' if not is_plural else 's'} with parameter{is_plural} passed in runtime:\n"
-                for key, values in conflicting_parameters.items():
-                    msg += f"`{key}`: config=`{values[0]}` | runtime=`{values[1]}`\n"
-                msg += """This happens, if you define values for pipeline parameters in configuration file and pass same parameters from the code. Example:
-```
-# config.yaml
-    parameters:
-        param_name: value1
-            
-            
-# pipeline.py
-@pipeline
-def pipeline_(param_name: str):
-    step_name()
-
-if __name__=="__main__":
-    pipeline_.with_options(config_path="config.yaml")(param_name="value2")
-```
-To avoid this consider setting pipeline parameters only in one place (config or code).
-"""
-                raise RuntimeError(msg)
-            for k, v_config in parameters_.items():
-                if k not in kwargs:
-                    kwargs[k] = v_config
-
-        try:
-            validated_args = pydantic_utils.validate_function_args(
-                self.entrypoint,
-                ConfigDict(arbitrary_types_allowed=False),
-                *args,
-                **kwargs,
+    def _prepare_invocations(self, **kwargs: Any) -> None:
+        """Prepares the invocations of the pipeline."""
+        for step in self._depends_on:
+            self.add_step_invocation(
+                step,
+                input_artifacts={},
+                external_artifacts={},
+                model_artifacts_or_metadata={},
+                client_lazy_loaders={},
+                parameters={},
+                default_parameters={},
+                upstream_steps=set(),
             )
-        except ValidationError as e:
-            raise ValueError(
-                "Invalid or missing pipeline function entrypoint arguments. "
-                "Only JSON serializable inputs are allowed as pipeline inputs. "
-                "Check out the pydantic error above for more details."
-            ) from e
-
-        self._parameters = validated_args
-        self._invocations = {}
-        with self:
-            for step in self._depends_on:
-                self.add_step_invocation(
-                    step,
-                    input_artifacts={},
-                    external_artifacts={},
-                    model_artifacts_or_metadata={},
-                    client_lazy_loaders={},
-                    parameters={},
-                    default_parameters={},
-                    upstream_steps=set(),
-                )
-
-    def add_dynamic_invocation(
-        self,
-        step: "BaseStep",
-        custom_id: Optional[str] = None,
-        allow_id_suffix: bool = True,
-        upstream_steps: Optional[Set[str]] = None,
-        input_artifacts: Dict[str, "StepArtifact"] = {},
-        external_artifacts: Dict[
-            str, Union[ExternalArtifact, "ArtifactVersionResponse"]
-        ] = {},
-    ) -> str:
-        """Adds a dynamic invocation to the pipeline.
-
-        Args:
-            step: The step for which to add an invocation.
-            custom_id: Custom ID to use for the invocation.
-            allow_id_suffix: Whether a suffix can be appended to the invocation
-                ID.
-            upstream_steps: The upstream steps for the invocation.
-            input_artifacts: The input artifacts for the invocation.
-            external_artifacts: The external artifacts for the invocation.
-
-        Returns:
-            The invocation ID.
-        """
-        invocation_id = self._compute_invocation_id(
-            step=step, custom_id=custom_id, allow_suffix=allow_id_suffix
-        )
-        invocation = StepInvocation(
-            id=invocation_id,
-            step=step,
-            input_artifacts=input_artifacts,
-            external_artifacts=external_artifacts,
-            model_artifacts_or_metadata={},
-            client_lazy_loaders={},
-            parameters=step.configuration.parameters,
-            default_parameters={},
-            upstream_steps=upstream_steps or set(),
-            pipeline=self,
-        )
-        self._invocations[invocation_id] = invocation
-        return invocation_id
 
     def __call__(
         self, *args: Any, **kwargs: Any
@@ -248,36 +146,6 @@ def __call__(
         self.prepare(*args, **kwargs)
         return self._run()
 
-    def _call_entrypoint(self, *args: Any, **kwargs: Any) -> None:
-        """Calls the pipeline entrypoint function with the given arguments.
-
-        Args:
-            *args: Entrypoint function arguments.
-            **kwargs: Entrypoint function keyword arguments.
-
-        Raises:
-            ValueError: If an input argument is missing or not JSON
-                serializable.
-        """
-        try:
-            validated_args = pydantic_utils.validate_function_args(
-                self.entrypoint,
-                ConfigDict(arbitrary_types_allowed=False),
-                *args,
-                **kwargs,
-            )
-        except ValidationError as e:
-            raise ValueError(
-                "Invalid or missing pipeline function entrypoint arguments. "
-                "Only JSON serializable inputs are allowed as pipeline inputs. "
-                "Check out the pydantic error above for more details."
-            ) from e
-
-        # Clear the invocations as they might still contain invocations from
-        # the compilation phase.
-        self._invocations = {}
-        self.entrypoint(**validated_args)
-
     def _compute_output_schema(self) -> Optional[Dict[str, Any]]:
         """Computes the output schema for the pipeline.