More fixes

Author: schustmi

src/zenml/config/compiler.py

@@ -489,40 +489,41 @@ def _compile_step_invocation(
         invocation.step = copy.deepcopy(invocation.step)
 
         step = invocation.step
-        if step_config:
-            step._apply_configuration(
-                step_config, runtime_parameters=invocation.parameters
-            )
+        with step._skip_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()
+            # Apply the dynamic configuration (which happened while executing the
+            # pipeline function) after all other step-specific configurations.
+            step._apply_dynamic_configuration()
 
-        convert_component_shortcut_settings_keys(
-            step.configuration.settings, stack=stack
-        )
-        step_spec = self._get_step_spec(invocation=invocation)
-        step_secrets = secret_utils.resolve_and_verify_secrets(
-            step.configuration.secrets
-        )
-        step_settings = self._filter_and_validate_settings(
-            settings=step.configuration.settings,
-            configuration_level=ConfigurationLevel.STEP,
-            stack=stack,
-        )
-        step.configure(
-            secrets=step_secrets,
-            settings=step_settings,
-            merge=False,
-        )
+            convert_component_shortcut_settings_keys(
+                step.configuration.settings, stack=stack
+            )
+            step_spec = self._get_step_spec(invocation=invocation)
+            step_secrets = secret_utils.resolve_and_verify_secrets(
+                step.configuration.secrets
+            )
+            step_settings = self._filter_and_validate_settings(
+                settings=step.configuration.settings,
+                configuration_level=ConfigurationLevel.STEP,
+                stack=stack,
+            )
+            step.configure(
+                secrets=step_secrets,
+                settings=step_settings,
+                merge=False,
+            )
 
-        parameters_to_ignore = (
-            set(step_config.parameters or {}) if step_config else set()
-        )
-        step_configuration_overrides = invocation.finalize(
-            parameters_to_ignore=parameters_to_ignore,
-            skip_input_validation=skip_input_validation,
-        )
+            parameters_to_ignore = (
+                set(step_config.parameters or {}) if step_config else set()
+            )
+            step_configuration_overrides = invocation.finalize(
+                parameters_to_ignore=parameters_to_ignore,
+                skip_input_validation=skip_input_validation,
+            )
         full_step_config = (
             step_configuration_overrides.apply_pipeline_configuration(
                 pipeline_configuration=pipeline_configuration

src/zenml/pipelines/dynamic/runner.py

@@ -318,7 +318,7 @@ def _await_and_validate_input(input: Any) -> Any:
 
     # TODO: we can validate the type of the inputs that are passed as raw data
     signature = inspect.signature(step.entrypoint, follow_wrapped=True)
-    validated_args = signature.bind(*args, **kwargs).arguments
+    validated_args = signature.bind_partial(*args, **kwargs).arguments
 
     return validated_args, upstream_steps
 
@@ -351,16 +351,18 @@ def _compile_step(
             update={"template": template.spec.invocation_id}
         )
 
-    # TODO:
-    # - differentiate between input artifacts and parameters?
-    # - default parameters
-    invocation_id = pipeline.add_dynamic_invocation(
+    _, _, _, _, _, default_parameters = step._parse_call_args(**inputs)
+    invocation_id = pipeline.add_step_invocation(
         step=step,
         custom_id=id,
         allow_id_suffix=not id,
         input_artifacts=input_artifacts,
         external_artifacts=external_artifacts,
         upstream_steps=upstream_steps,
+        parameters={},
+        default_parameters=default_parameters,
+        model_artifacts_or_metadata={},
+        client_lazy_loaders={},
     )
     compiled_step = Compiler()._compile_step_invocation(
         invocation=pipeline.invocations[invocation_id],

src/zenml/pipelines/pipeline_definition.py

@@ -1361,7 +1361,7 @@ def add_step_invocation(
         Returns:
             The step invocation ID.
         """
-        if Pipeline.ACTIVE_PIPELINE != self:
+        if not self.is_dynamic and Pipeline.ACTIVE_PIPELINE != self:
             raise RuntimeError(
                 "A step invocation can only be added to an active pipeline."
             )

src/zenml/steps/base_step.py

@@ -18,10 +18,12 @@
 import inspect
 from abc import abstractmethod
 from collections import defaultdict
+from contextlib import contextmanager
 from typing import (
     TYPE_CHECKING,
     Any,
     Dict,
+    Generator,
     List,
     Mapping,
     Optional,
@@ -207,6 +209,7 @@ def __init__(
 
         self._configuration = PartialStepConfiguration(name=name)
         self._dynamic_configuration: Optional["StepConfigurationUpdate"] = None
+        self._should_skip_dynamic_configuration = False
         self.configure(
             enable_cache=enable_cache,
             enable_artifact_metadata=enable_artifact_metadata,
@@ -492,7 +495,9 @@ def __call__(
         if context := DynamicPipelineRunContext.get():
             after = cast(
                 Union[
-                    "StepRunOutputsFuture", Sequence["StepRunOutputsFuture"], None
+                    "StepRunOutputsFuture",
+                    Sequence["StepRunOutputsFuture"],
+                    None,
                 ],
                 after,
             )
@@ -910,6 +915,13 @@ def copy(self) -> "BaseStep":
 
         return copy_
 
+    @contextmanager
+    def _skip_dynamic_configuration(self) -> Generator[None, None, None]:
+        """Context manager to skip the dynamic configuration."""
+        self._should_skip_dynamic_configuration = True
+        yield
+        self._should_skip_dynamic_configuration = False
+
     def _apply_configuration(
         self,
         config: "StepConfigurationUpdate",
@@ -929,7 +941,10 @@ def _apply_configuration(
 
         self._validate_configuration(config, runtime_parameters)
 
-        if DynamicPipelineRunContext.get():
+        if (
+            DynamicPipelineRunContext.get()
+            and not self._should_skip_dynamic_configuration
+        ):
             if self._dynamic_configuration is None:
                 self._dynamic_configuration = config
             else: