Docs

Author: schustmi

docs/book/how-to/steps-pipelines/dynamic_pipelines.md

@@ -250,6 +250,17 @@ if __name__ == "__main__":
 
 The `depends_on` parameter tells ZenML which steps can be configured via the YAML file. This is particularly useful when you want to allow users to configure pipeline behavior without modifying code.
 
+### Pass pipeline parameters when running snapshots from the server
+
+When running a snapshot from the server (either via the UI or the SDK/Rest API), you can now pass pipeline parameters for your dynamic pipelines.
+
+For example:
+```python
+from zenml.client import Client
+
+Client().trigger_pipeline(snapshot_id=<ID>, run_configuration={"parameters": {"my_param": 3}})
+```
+
 ## Limitations and Known Issues
 
 ### Logging

src/zenml/zen_stores/template_utils.py

@@ -236,7 +236,8 @@ def generate_config_schema(
     all_steps: Dict[str, Any] = {}
     all_steps_required = False
     for step_name, step in step_configurations.items():
-        step_fields = generic_step_fields.copy()
+        step_fields: Dict[str, Any] = {}
+
         if step.config.parameters:
             parameter_fields: Dict[str, Any] = {}
 
@@ -259,6 +260,7 @@ def generate_config_schema(
                 FieldInfo(default=...),
             )
 
+        step_fields.update(generic_step_fields)
         step_model = create_model(step_name, **step_fields)
 
         # Pydantic doesn't allow field names to start with an underscore
@@ -285,32 +287,13 @@ def generate_config_schema(
 
     top_level_fields: Dict[str, Any] = {}
 
-    for key, field_info in PipelineRunConfiguration.model_fields.items():
-        if key in ["schedule", "build", "steps", "settings", "parameters"]:
-            continue
-
-        if field_info.annotation == Optional[SourceWithValidator]:  # type: ignore[comparison-overlap]
-            top_level_fields[key] = (Optional[str], field_info)
-        else:
-            top_level_fields[key] = (field_info.annotation, field_info)
-
-    top_level_fields["settings"] = (Optional[settings_model], None)
-
-    if all_steps_required:
-        top_level_fields["steps"] = (all_steps_model, FieldInfo(default=...))
-    else:
-        top_level_fields["steps"] = (
-            Optional[all_steps_model],
-            FieldInfo(default=None),
-        )
-
     if snapshot.is_dynamic:
         pipeline_parameter_fields: Dict[str, Any] = {}
 
         for parameter_name in pipeline_configuration.parameters or {}:
             # Pydantic doesn't allow field names to start with an underscore
             sanitized_parameter_name = parameter_name.lstrip("_")
-            while sanitized_parameter_name in parameter_fields:
+            while sanitized_parameter_name in pipeline_parameter_fields:
                 sanitized_parameter_name = sanitized_parameter_name + "_"
 
             pipeline_parameter_fields[sanitized_parameter_name] = (
@@ -326,4 +309,23 @@ def generate_config_schema(
             FieldInfo(default=None),
         )
 
+    for key, field_info in PipelineRunConfiguration.model_fields.items():
+        if key in ["schedule", "build", "steps", "settings", "parameters"]:
+            continue
+
+        if field_info.annotation == Optional[SourceWithValidator]:  # type: ignore[comparison-overlap]
+            top_level_fields[key] = (Optional[str], field_info)
+        else:
+            top_level_fields[key] = (field_info.annotation, field_info)
+
+    top_level_fields["settings"] = (Optional[settings_model], None)
+
+    if all_steps_required:
+        top_level_fields["steps"] = (all_steps_model, FieldInfo(default=...))
+    else:
+        top_level_fields["steps"] = (
+            Optional[all_steps_model],
+            FieldInfo(default=None),
+        )
+
     return create_model("Result", **top_level_fields).model_json_schema()  # type: ignore[no-any-return]