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

Author: stefannica

src/zenml/config/compiler.py

@@ -656,14 +656,10 @@ def _compute_pipeline_spec(
             logger.warning("Failed to compute pipeline output schema: %s", e)
             output_schema = None
 
-        try:
-            parameters_model = pipeline.get_parameters_model()
-            if parameters_model:
-                input_schema = parameters_model.model_json_schema()
-            else:
-                input_schema = None
-        except Exception as e:
-            logger.warning("Failed to compute pipeline input schema: %s", e)
+        parameters_model = pipeline.get_parameters_model()
+        if parameters_model:
+            input_schema = parameters_model.model_json_schema()
+        else:
             input_schema = None
 
         return PipelineSpec(

src/zenml/deployers/server/app.py

@@ -71,7 +71,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 
     try:
         global _service
-        # Defer UUID parsing to the service itself to simplify testing
         _service = PipelineDeploymentService(snapshot_id)
         _service.initialize()
         # params model is available.

src/zenml/deployers/server/parameters.py

@@ -11,15 +11,7 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 #  or implied. See the License for the specific language governing
 #  permissions and limitations under the License.
-"""Shared utilities to construct and validate pipeline parameter models.
-
-This module centralizes the logic to:
-- Build a Pydantic model for pipeline parameters from a snapshot
-- Validate and normalize request parameters using that model
-
-It is intentionally independent of FastAPI or deployment internals so that
-other entry points (e.g., CLI) can reuse the same behavior.
-"""
+"""Parameters model builder for deployments."""
 
 from typing import Optional, Type
 
@@ -34,9 +26,8 @@
 
 
 def build_params_model_from_snapshot(
-    snapshot: PipelineSnapshotResponse,
     *,
-    strict: bool = True,
+    snapshot: PipelineSnapshotResponse,
 ) -> Optional[Type[BaseModel]]:
     """Construct a Pydantic model representing pipeline parameters.
 
@@ -61,8 +52,7 @@ def build_params_model_from_snapshot(
             f"Snapshot `{snapshot.id}` is missing pipeline_spec.source; "
             "cannot build parameter model."
         )
-        if strict:
-            raise RuntimeError(msg)
+        logger.error(msg)
         return None
 
     try:
@@ -71,19 +61,18 @@ def build_params_model_from_snapshot(
         )
     except Exception as e:
         logger.debug(f"Failed to load pipeline class from snapshot: {e}")
-        if strict:
-            raise
-        return None
+        logger.error(f"Failed to load pipeline class from snapshot: {e}")
+        raise RuntimeError(f"Failed to load pipeline class from snapshot: {e}")
 
     model = pipeline_class.get_parameters_model()
     if not model:
         message = (
             f"Failed to construct parameters model from pipeline "
             f"`{snapshot.pipeline_configuration.name}`."
         )
-        if strict:
-            raise RuntimeError(message)
-        else:
+        logger.error(message)
+        raise RuntimeError(message)
+    else:
             logger.debug(message)
 
     return model

src/zenml/deployers/server/runtime.py

@@ -39,7 +39,6 @@ class _DeploymentState(BaseModel):
     snapshot_id: Optional[str] = None
     pipeline_parameters: Dict[str, Any] = Field(default_factory=dict)
     outputs: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
-    # Per-request in-memory mode override
 
     # In-memory data storage for artifacts
     in_memory_data: Dict[str, Any] = Field(default_factory=dict)
@@ -55,7 +54,6 @@ def reset(self) -> None:
         self.in_memory_data.clear()
 
 
-# Use contextvars for thread-safe, request-scoped state
 _deployment_context: contextvars.ContextVar[_DeploymentState] = (
     contextvars.ContextVar("deployment_context", default=_DeploymentState())
 )
@@ -169,19 +167,4 @@ def get_in_memory_data(uri: str) -> Any:
     if is_active():
         state = _get_context()
         return state.in_memory_data.get(uri)
-    return None
-
-
-def has_in_memory_data(uri: str) -> bool:
-    """Check if data exists in memory for the given URI.
-
-    Args:
-        uri: The artifact URI to check.
-
-    Returns:
-        True if data exists in memory for the URI.
-    """
-    if is_active():
-        state = _get_context()
-        return uri in state.in_memory_data
-    return False
+    return None

src/zenml/deployers/server/service.py

@@ -34,6 +34,7 @@
     ServiceInfo,
     SnapshotInfo,
 )
+from zenml.deployers.server.parameters import build_params_model_from_snapshot
 from zenml.enums import StackComponentType
 from zenml.hooks.hook_validators import load_and_run_hook
 from zenml.integrations.registry import integration_registry
@@ -135,7 +136,7 @@ def initialize(self) -> None:
             integration_registry.activate_integrations()
 
             # Build parameter model
-            self._params_model = self._build_params_model()
+            self._params_model = build_params_model_from_snapshot(self.snapshot, strict=True)
 
             # Initialize orchestrator
             self._orchestrator = SharedLocalOrchestrator(
@@ -324,14 +325,6 @@ def _map_outputs(
 
         return filtered_outputs
 
-    def _prepare_execute_with_orchestrator(
-        self,
-    ) -> PipelineRunResponse:
-        # Create a placeholder run and execute with a known run id
-        return run_utils.create_placeholder_run(
-            snapshot=self.snapshot, logs=None
-        )
-
     def _execute_with_orchestrator(
         self,
         placeholder_run: PipelineRunResponse,
@@ -411,25 +404,6 @@ def _execute_with_orchestrator(
         # Store captured outputs for the caller to use
         return captured_outputs
 
-    def _build_params_model(self) -> Any:
-        """Build the pipeline parameters model from the deployment.
-
-        Returns:
-            A parameters model derived from the deployment configuration.
-
-        Raises:
-            Exception: If the model cannot be constructed.
-        """
-        try:
-            from zenml.deployers.server.parameters import (
-                build_params_model_from_snapshot,
-            )
-
-            return build_params_model_from_snapshot(self.snapshot, strict=True)
-        except Exception as e:
-            logger.error(f"Failed to construct parameter model: {e}")
-            raise
-
     def _execute_init_hook(self) -> None:
         """Execute init hook if present.
 

src/zenml/orchestrators/step_runner.py

@@ -142,8 +142,6 @@ def run(
         Raises:
             BaseException: A general exception if the step fails.
         """
-        # Store step_run_info for effective config access
-        self._step_run_info = step_run_info
         if handle_bool_env_var(ENV_ZENML_DISABLE_STEP_LOGS_STORAGE, False):
             step_logging_enabled = False
         else:
@@ -415,7 +413,7 @@ def _load_step(self) -> "BaseStep":
 
         step_instance = BaseStep.load_from_source(self._step.spec.source)
         step_instance = copy.deepcopy(step_instance)
-        step_instance._configuration = self._step_run_info.config
+        step_instance._configuration = self._step.config
         return step_instance
 
     def _load_output_materializers(

src/zenml/orchestrators/utils.py

@@ -434,8 +434,6 @@ def deployment_snapshot_request_from_source_snapshot(
 
     if source_snapshot.stack is None:
         raise ValueError("Source snapshot stack is None")
-    if source_snapshot.pipeline is None:
-        raise ValueError("Source snapshot pipeline is None")
 
     return PipelineSnapshotRequest(
         project=source_snapshot.project_id,

tests/integration/functional/deployers/serving/test_app_endpoints.py

@@ -326,30 +326,6 @@ def test_invoke_endpoint_execution_failure(
         assert response.status_code == 200
         assert response.json()["success"] is False
 
-    def test_invoke_requires_auth_when_enabled(
-        self,
-        client_service_pair: Tuple[
-            TestClient, StubPipelineServingService, ModuleType
-        ],
-        monkeypatch: pytest.MonkeyPatch,
-    ) -> None:
-        """Enforce authentication when the auth key is configured."""
-        client, _, _ = client_service_pair
-        monkeypatch.setenv("ZENML_DEPLOYMENT_AUTH_KEY", "secret")
-
-        response = client.post(
-            "/invoke", json={"parameters": {"city": "Paris"}}
-        )
-        assert response.status_code == 401
-
-        response = client.post(
-            "/invoke",
-            json={"parameters": {"city": "Paris"}},
-            headers={"Authorization": "Bearer secret"},
-        )
-        assert response.status_code == 200
-        monkeypatch.delenv("ZENML_DEPLOYMENT_AUTH_KEY")
-
     def test_cleanup_called_on_shutdown(
         self,
         monkeypatch: pytest.MonkeyPatch,

tests/unit/deployers/serving/test_runtime.py

@@ -160,19 +160,13 @@ def test_in_memory_data_storage(self):
         )
         assert runtime.get_in_memory_data("memory://missing") is None
 
-        # Check existence
-        assert runtime.has_in_memory_data("memory://artifact/1")
-        assert runtime.has_in_memory_data("memory://artifact/2")
-        assert not runtime.has_in_memory_data("memory://missing")
-
     def test_in_memory_data_inactive_context(self):
         """Test in-memory data operations when context is inactive."""
         # Don't start context
         runtime.put_in_memory_data("memory://artifact/1", {"data": "value"})
 
         # Should not store anything
         assert runtime.get_in_memory_data("memory://artifact/1") is None
-        assert not runtime.has_in_memory_data("memory://artifact/1")
 
     def test_context_isolation(self):
         """Test that multiple contexts don't interfere with each other."""
@@ -274,7 +268,6 @@ def test_context_reset_clears_all_data(self):
         # Verify data is stored
         assert runtime.is_active()
         assert runtime.get_outputs() != {}
-        assert runtime.has_in_memory_data("memory://artifact/1")
         assert runtime.should_use_in_memory_mode() is True
 
         # Stop context (triggers reset)
@@ -292,5 +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 not runtime.has_in_memory_data("memory://artifact/1")
         assert runtime.should_use_in_memory_mode() is False