Fix inconsistent key names in PATCH pipeline endpoint (#5089)

Author: itallix

application/backend/app/api/routers/pipelines.py

@@ -20,22 +20,21 @@
 router = APIRouter(prefix="/api/projects/{project_id}/pipeline", tags=["Pipelines"])
 
 UPDATE_PIPELINE_BODY_DESCRIPTION = """
-Partial pipeline configuration update. May contain any subset of fields including 'name', 'source_id', 
-'sink_id', or 'model_id'. Fields not included in the request will remain unchanged.
+Partial pipeline configuration update. May contain any subset of fields including 'device', 'data_collection_policies', 
+'source_id', 'sink_id', or 'model_id'. Fields not included in the request will remain unchanged.
 """
 UPDATE_PIPELINE_BODY_EXAMPLES = {
     "switch_model": Example(
         summary="Switch active model",
-        description="Change the active model for the pipeline",
+        description="Change the active model of the pipeline",
         value={
             "model_id": "c1feaabc-da2b-442e-9b3e-55c11c2c2ff3",
         },
     ),
     "reconfigure": Example(
         summary="Reconfigure pipeline",
-        description="Change the name, source and sink of the pipeline",
+        description="Change the source and the sink of the pipeline",
         value={
-            "name": "Updated Production Pipeline",
             "source_id": "e3cbd8d0-17b8-463e-85a2-4aaed031674e",
             "sink_id": "c6787c06-964b-4097-8eca-238b8cf79fc9",
         },

application/backend/app/models/pipeline.py

@@ -5,7 +5,7 @@
 from typing import Any
 from uuid import UUID
 
-from pydantic import Field, model_validator
+from pydantic import AliasChoices, Field, model_validator
 
 from .base import BaseEntity
 from .data_collection_policy import DataCollectionPolicy
@@ -41,7 +41,7 @@ class Pipeline(BaseEntity):
         model_revision: The model revision to use for processing, None if no model is selected.
         source_id: UUID reference to the source entity.
         sink_id: UUID reference to the sink entity.
-        model_revision_id: UUID reference to the model revision entity.
+        model_id: UUID reference to the model revision entity.
         status: Current operational status of the pipeline (IDLE or RUNNING).
         data_collection_policies: List of policies governing data collection behavior during pipeline execution.
         device: The device used for model inference (e.g., 'cpu', 'xpu', 'cuda', 'xpu-1', etc.).
@@ -56,7 +56,7 @@ class Pipeline(BaseEntity):
     model_revision: ModelRevision | None = None
     source_id: UUID | None = None
     sink_id: UUID | None = None
-    model_revision_id: UUID | None = None
+    model_id: UUID | None = Field(default=None, validation_alias=AliasChoices("model_revision_id", "model_id"))
     status: PipelineStatus = PipelineStatus.IDLE
     data_collection_policies: list[DataCollectionPolicy] = Field(default_factory=list)
     device: str = Field(default="cpu", pattern=r"^(cpu|xpu|cuda)(-\d+)?$")
@@ -73,7 +73,7 @@ def set_status_from_is_running(cls, data: Any) -> Any:
     @model_validator(mode="after")
     def validate_running_status(self) -> "Pipeline":
         if self.status == PipelineStatus.RUNNING and any(
-            x is None for x in (self.source_id, self.sink_id, self.model_revision_id)
+            x is None for x in (self.source_id, self.sink_id, self.model_id)
         ):
             raise ValueError("Pipeline cannot be in 'running' state when source, sink, or model is not configured.")
         return self

application/backend/app/services/event/event_bus.py

@@ -10,6 +10,7 @@
 class EventType(StrEnum):
     SOURCE_CHANGED = "SOURCE_CHANGED"
     SINK_CHANGED = "SINK_CHANGED"
+    MODEL_CHANGED = "MODEL_CHANGED"
     PIPELINE_DATASET_COLLECTION_POLICIES_CHANGED = "PIPELINE_DATASET_COLLECTION_POLICIES_CHANGED"
     PIPELINE_STATUS_CHANGED = "PIPELINE_STATUS_CHANGED"
     INFERENCE_DEVICE_CHANGED = "INFERENCE_DEVICE_CHANGED"
@@ -51,6 +52,9 @@ def _should_notify_source(self, event_type: EventType) -> bool:
     def _should_notify_sink(self, event_type: EventType) -> bool:
         return event_type in (EventType.SINK_CHANGED, EventType.PIPELINE_STATUS_CHANGED)
 
+    def _should_notify_model(self, event_type: EventType) -> bool:
+        return event_type in (EventType.MODEL_CHANGED, EventType.PIPELINE_STATUS_CHANGED)
+
     def emit_event(self, event_type: EventType) -> None:
         super().emit_event(event_type)
 
@@ -60,5 +64,5 @@ def emit_event(self, event_type: EventType) -> None:
         if self._should_notify_sink(event_type):
             self._notify_all(self._sink_changed_condition)
 
-        if event_type == EventType.PIPELINE_STATUS_CHANGED and self._model_reload_event:
+        if self._should_notify_model(event_type) and self._model_reload_event:
             self._model_reload_event.set()

application/backend/app/services/pipeline_metrics_service.py

@@ -36,7 +36,7 @@ def get_pipeline_metrics(self, pipeline_id: UUID, time_window: int = 60) -> Pipe
 
         # Get actual latency measurements from the metrics service
         latency_samples = self._metrics_service.get_latency_measurements(
-            model_id=pipeline.model_revision_id,  # type: ignore[arg-type] # model is always there for running pipeline
+            model_id=pipeline.model_id,  # type: ignore[arg-type] # model is always there for running pipeline
             time_window=time_window,
         )
 
@@ -55,7 +55,7 @@ def get_pipeline_metrics(self, pipeline_id: UUID, time_window: int = 60) -> Pipe
 
         # Get throughput measurements from the metrics service
         total_requests, throughput_data = self._metrics_service.get_throughput_measurements(
-            model_id=pipeline.model_revision_id,  # type: ignore[arg-type]
+            model_id=pipeline.model_id,  # type: ignore[arg-type]
             time_window=time_window,
         )
         if total_requests:

application/backend/app/services/pipeline_service.py

@@ -51,13 +51,14 @@ def is_running(self, project_id: UUID) -> bool:
     def update_pipeline(self, project_id: UUID, partial_config: dict) -> Pipeline:
         """Update an existing pipeline."""
         pipeline = self.get_pipeline_by_id(project_id)
-        to_update = type(pipeline).model_validate(pipeline.model_copy(update=partial_config))
+        base = pipeline.model_dump()
+        to_update = type(pipeline).model_validate({**base, **partial_config})
         pipeline_repo = PipelineRepository(self._db_session)
         to_update_db = PipelineDB(
             project_id=str(to_update.project_id),
             source_id=str(to_update.source_id) if to_update.source_id else None,
             sink_id=str(to_update.sink_id) if to_update.sink_id else None,
-            model_revision_id=str(to_update.model_revision_id) if to_update.model_revision_id else None,
+            model_revision_id=str(to_update.model_id) if to_update.model_id else None,
             is_running=to_update.status.as_bool,
             data_collection_policies=[obj.model_dump() for obj in to_update.data_collection_policies],
             device=to_update.device,
@@ -74,6 +75,8 @@ def update_pipeline(self, project_id: UUID, partial_config: dict) -> Pipeline:
                 self._event_bus.emit_event(EventType.PIPELINE_DATASET_COLLECTION_POLICIES_CHANGED)
             if pipeline.device != updated.device:
                 self._event_bus.emit_event(EventType.INFERENCE_DEVICE_CHANGED)
+            if pipeline.model_id != updated.model_revision.id:  # type: ignore[union-attr] # model_revision is always there for running pipeline
+                self._event_bus.emit_event(EventType.MODEL_CHANGED)
         elif pipeline.status != updated.status:
             # If the pipeline is being activated or stopped
             self._event_bus.emit_event(EventType.PIPELINE_STATUS_CHANGED)

application/backend/tests/conftest.py

@@ -77,7 +77,7 @@ def fxt_running_pipeline(fxt_webcam_source, fxt_mqtt_sink, fxt_model) -> Pipelin
         project_id=uuid4(),
         source_id=fxt_webcam_source.id,
         sink_id=fxt_mqtt_sink.id,
-        model_revision_id=fxt_model.id,
+        model_id=fxt_model.id,
         status=PipelineStatus.RUNNING,
     )
 

application/backend/tests/integration/services/test_dataset_service.py

@@ -493,7 +493,7 @@ def test_create_dataset_item(
             data=image,
             user_reviewed=user_reviewed,
             source_id=pipeline.source_id if use_pipeline_source else None,
-            prediction_model_id=pipeline.model_revision_id if use_pipeline_model else None,
+            prediction_model_id=pipeline.model_id if use_pipeline_model else None,
             annotations=fxt_annotations(label_id) if not user_reviewed else None,
         )
 
@@ -515,7 +515,7 @@ def test_create_dataset_item(
         else:
             assert dataset_item.source_id is None
         if use_pipeline_model:
-            assert dataset_item.prediction_model_id == str(pipeline.model_revision_id)
+            assert dataset_item.prediction_model_id == str(pipeline.model_id)
         else:
             assert dataset_item.prediction_model_id is None
         if not user_reviewed:

application/backend/tests/integration/services/test_pipeline_service.py

@@ -72,7 +72,7 @@ def test_create_pipeline(self, fxt_pipeline_service, fxt_project_id, fxt_db_proj
             and pipeline.source is None
             and pipeline.sink_id is None
             and pipeline.sink is None
-            and pipeline.model_revision_id is None
+            and pipeline.model_id is None
             and pipeline.model_revision is None
             and pipeline.status == PipelineStatus.IDLE
             and pipeline.data_collection_policies == []
@@ -91,7 +91,7 @@ def test_get_pipeline(self, fxt_pipeline_service, fxt_project_id, fxt_project_wi
         assert pipeline.status == PipelineStatus.IDLE
         assert pipeline.sink.name == db_pipeline.sink.name
         assert pipeline.source.name == db_pipeline.source.name
-        assert str(pipeline.model_revision_id) == db_pipeline.model_revision_id
+        assert str(pipeline.model_id) == db_pipeline.model_revision_id
         assert pipeline.data_collection_policies == [FixedRateDataCollectionPolicy(rate=0.1)]
 
     def test_get_active_pipeline(self, fxt_pipeline_service, fxt_project_with_pipeline, db_session):
@@ -142,6 +142,27 @@ def test_reconfigure_running_pipeline(
         assert str(getattr(updated, pipeline_attr)) == item_id
         assert str(getattr(updated, pipeline_attr)) == getattr(db_updated, pipeline_attr)
 
+    @pytest.mark.parametrize("model_attr", ["model_id", "model_revision_id"])
+    def test_switch_model(
+        self,
+        model_attr,
+        fxt_project_with_pipeline,
+        fxt_db_models,
+        fxt_pipeline_service,
+        fxt_event_bus,
+        db_session,
+    ):
+        """Test updating a pipeline by ID."""
+        _, db_pipeline = fxt_project_with_pipeline(is_running=True)
+
+        model_id = fxt_db_models[1].id
+        updated = fxt_pipeline_service.update_pipeline(db_pipeline.project_id, {model_attr: model_id})
+
+        fxt_event_bus.emit_event.assert_called_once_with(EventType.MODEL_CHANGED)
+        db_updated = db_session.get(PipelineDB, db_pipeline.project_id)
+        assert str(updated.model_id) == model_id
+        assert str(updated.model_id) == db_updated.model_revision_id
+
     @pytest.mark.parametrize("pipeline_status", [PipelineStatus.IDLE, PipelineStatus.RUNNING])
     def test_enable_disable_pipeline(
         self,

application/backend/tests/unit/services/event/test_event_bus.py

@@ -69,6 +69,18 @@ def test_sink_changed(self, fxt_event_bus: EventBusFactory) -> None:
             notified = sink_changed_condition.acquire()
         assert notified
 
+    def test_model_changed(self, fxt_event_bus: EventBusFactory) -> None:
+        """Test model changed"""
+        handler = MagicMock(spec=Callable)
+        model_reload_event = mp.Event()
+        event_bus = fxt_event_bus(None, None, model_reload_event)
+        event_bus.subscribe(event_types=[EventType.MODEL_CHANGED], handler=handler)
+
+        event_bus.emit_event(EventType.MODEL_CHANGED)
+
+        handler.assert_called_once_with()
+        assert model_reload_event.is_set()
+
     def test_pipeline_dataset_collection_policies_changed(self, fxt_event_bus: EventBusFactory) -> None:
         """Test pipeline dataset collection policies changed"""
         handler = MagicMock(spec=Callable)

application/docs/models.md

@@ -62,7 +62,7 @@ The lifecycle of a model revision in Geti Tune consists of several key stages:
 
 When users want to create a new model revision, they can choose any architecture compatible with the task, as well as
 the base model to fine-tune from. Such weights can be either from a public pre-trained model or from an existing
-model revision in the same project. In the latter case, an parent-child relationship is established between the two
+model revision in the same project. In the latter case, a parent-child relationship is established between the two
 model revisions; these links form a versioning chain that allows users to track the evolution of models over time.
 Revisions trained from scratch, namely from the pre-trained weights, do not have a parent model. A model revision may
 be used as a base for multiple new model revisions, allowing users to experiment with different training configurations.