Merge branch 'feature/served-pipelines' of github.com:zenml-io/zenml into feature/served-pipelines

Author: stefannica

src/zenml/deployers/server/runtime.py

@@ -34,7 +34,7 @@ class _DeploymentState(BaseModel):
     model_config = {"extra": "forbid"}
 
     active: bool = False
-    use_in_memory: bool = False
+    skip_artifact_materialization: bool = False
     request_id: Optional[str] = None
     snapshot_id: Optional[str] = None
     pipeline_parameters: Dict[str, Any] = Field(default_factory=dict)
@@ -50,7 +50,7 @@ def reset(self) -> None:
         self.snapshot_id = None
         self.pipeline_parameters.clear()
         self.outputs.clear()
-        self.use_in_memory = False
+        self.skip_artifact_materialization = False
         self.in_memory_data.clear()
 
 
@@ -72,23 +72,23 @@ def start(
     request_id: str,
     snapshot: PipelineSnapshotResponse,
     parameters: Dict[str, Any],
-    use_in_memory: bool = False,
+    skip_artifact_materialization: bool = False,
 ) -> None:
     """Initialize deployment state for the current request context.
 
     Args:
         request_id: The ID of the request.
         snapshot: The snapshot to deploy.
         parameters: The parameters to deploy.
-        use_in_memory: Whether to use in-memory mode.
+        skip_artifact_materialization: Whether to skip artifact materialization.
     """
     state = _DeploymentState()
     state.active = True
     state.request_id = request_id
     state.snapshot_id = str(snapshot.id)
     state.pipeline_parameters = parameters
     state.outputs = {}
-    state.use_in_memory = use_in_memory
+    state.skip_artifact_materialization = skip_artifact_materialization
     _deployment_context.set(state)
 
 
@@ -131,15 +131,15 @@ def get_outputs() -> Dict[str, Dict[str, Any]]:
     return dict(_get_context().outputs)
 
 
-def should_use_in_memory_mode() -> bool:
-    """Check if the current request should use in-memory mode.
+def should_skip_artifact_materialization() -> bool:
+    """Check if the current request should skip artifact materialization.
 
     Returns:
-        True if in-memory mode is enabled for this request.
+        True if artifact materialization is skipped for this request.
     """
     if is_active():
         state = _get_context()
-        return state.use_in_memory
+        return state.skip_artifact_materialization
     return False
 
 

src/zenml/deployers/server/service.py

@@ -398,7 +398,7 @@ def _execute_with_orchestrator(
             request_id=str(uuid4()),
             snapshot=deployment_snapshot,
             parameters=resolved_params,
-            use_in_memory=use_in_memory,
+            skip_artifact_materialization=use_in_memory,
         )
 
         captured_outputs: Optional[Dict[str, Dict[str, Any]]] = None

src/zenml/orchestrators/step_launcher.py

@@ -442,7 +442,7 @@ def _run_step(
             step_run=step_run,
             stack=self._stack,
             step=self._step,
-            create_dirs=not runtime.should_use_in_memory_mode(),
+            create_dirs=not runtime.should_skip_artifact_materialization(),
         )
 
         start_time = time.time()

src/zenml/orchestrators/step_runner.py

@@ -673,7 +673,7 @@ def _store_output_artifacts(
                 materializer_class = materializer_registry[data_type]
 
             # Choose materializer class upfront based on serving mode
-            if runtime.should_use_in_memory_mode():
+            if runtime.should_skip_artifact_materialization():
                 materializer_class = InMemoryMaterializer
             uri = output_artifact_uris[output_name]
             artifact_config = output_annotations[output_name].artifact_config

src/zenml/pipelines/pipeline_definition.py

@@ -1808,7 +1808,14 @@ def _compute_input_model(self) -> Optional[Type[BaseModel]]:
             defaults: Dict[str, Any] = self._parameters
             model_args: Dict[str, Any] = {}
             for name, param in entrypoint_definition.inputs.items():
-                model_args[name] = (param.annotation, defaults.get(name, ...))
+                if name in defaults:
+                    default_value = defaults[name]
+                elif param.default is not inspect.Parameter.empty:
+                    default_value = param.default
+                else:
+                    default_value = ...
+
+                model_args[name] = (param.annotation, default_value)
 
             model_args["__config__"] = ConfigDict(extra="forbid")
             params_model: Type[BaseModel] = create_model(

tests/unit/deployers/server/test_runtime.py

@@ -219,8 +219,8 @@ def test_use_in_memory_setting(self):
             use_in_memory=True,
         )
 
-        assert runtime.should_use_in_memory_mode() is True
-        assert runtime.should_use_in_memory_mode() is True
+        assert runtime.should_skip_artifact_materialization() is True
+        assert runtime.should_skip_artifact_materialization() is True
 
         runtime.stop()
 
@@ -232,7 +232,7 @@ def test_use_in_memory_setting(self):
             use_in_memory=False,
         )
 
-        assert runtime.should_use_in_memory_mode() is False
+        assert runtime.should_skip_artifact_materialization() is False
 
         runtime.stop()
 
@@ -243,11 +243,11 @@ def test_use_in_memory_setting(self):
             parameters={},
         )
 
-        assert runtime.should_use_in_memory_mode() is False
+        assert runtime.should_skip_artifact_materialization() is False
 
     def test_use_in_memory_inactive_context(self):
         """Test use_in_memory functions when context is inactive."""
-        assert runtime.should_use_in_memory_mode() is False
+        assert runtime.should_skip_artifact_materialization() is False
 
     def test_context_reset_clears_all_data(self):
         """Test that context reset clears all stored data."""
@@ -268,7 +268,7 @@ def test_context_reset_clears_all_data(self):
         # Verify data is stored
         assert runtime.is_active()
         assert runtime.get_outputs() != {}
-        assert runtime.should_use_in_memory_mode() is True
+        assert runtime.should_skip_artifact_materialization() is True
 
         # Stop context (triggers reset)
         runtime.stop()
@@ -285,4 +285,4 @@ def test_context_reset_clears_all_data(self):
 
         assert runtime.get_outputs() == {}
         assert runtime.get_in_memory_data("memory://artifact/1") is None
-        assert runtime.should_use_in_memory_mode() is False
+        assert runtime.should_skip_artifact_materialization() is False

tests/unit/materializers/test_in_memory_materializer.py

@@ -14,7 +14,9 @@ def test_in_memory_materializer_uses_runtime(monkeypatch) -> None:
     from zenml.deployers.server import runtime
 
     monkeypatch.setattr(runtime, "is_active", lambda: True)
-    monkeypatch.setattr(runtime, "should_use_in_memory_mode", lambda: True)
+    monkeypatch.setattr(
+        runtime, "should_skip_artifact_materialization", lambda: True
+    )
     monkeypatch.setattr(runtime, "put_in_memory_data", stored.__setitem__)
     monkeypatch.setattr(runtime, "get_in_memory_data", stored.get)