UN-151 [FIX] Block workflow deletion at backend when in use by pipelines/APIs

backend/workflow_manager/workflow_v2/exceptions.py

@@ -21,6 +21,11 @@ class WorkflowDoesNotExistError(APIException):
     default_detail = "Workflow does not exist"
 
 
+class WorkflowDeletionError(APIException):
+    status_code = 400
+    default_detail = "Workflow cannot be deleted as it is currently in use."
+
+
 class ExecutionDoesNotExistError(APIException):
     status_code = 404
     default_detail = "Execution does not exist."

backend/workflow_manager/workflow_v2/views.py

@@ -40,6 +40,7 @@
 from workflow_manager.workflow_v2.enums import SchemaEntity, SchemaType
 from workflow_manager.workflow_v2.exceptions import (
     InternalException,
+    WorkflowDeletionError,
     WorkflowDoesNotExistError,
     WorkflowGenerationError,
     WorkflowRegenerationError,
@@ -136,6 +137,23 @@ def perform_create(self, serializer: WorkflowSerializer) -> Workflow:
             raise WorkflowGenerationError
         return workflow
 
+    def perform_destroy(self, instance: Workflow) -> None:
+        """Block deletion when the workflow is in use by any pipeline/API.
+
+        The frontend gates the delete button via `can_update`, but direct API
+        callers can still hit DELETE. Without this guard, the CASCADE FK on
+        Pipeline/APIDeployment would silently drop their rows along with
+        the workflow.
+        """
+        usage = WorkflowHelper.can_update_workflow(str(instance.id))
+        if not usage.get("can_update", False):
+            raise WorkflowDeletionError(
+                detail=WorkflowHelper.build_workflow_in_use_message(
+                    instance.workflow_name, usage
+                )
+            )
+        super().perform_destroy(instance)
+
     def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         """Override partial_update to handle sharing notifications."""
         # Get the workflow instance before update

backend/workflow_manager/workflow_v2/workflow_helper.py

@@ -991,6 +991,46 @@ def make_async_result(obj: AsyncResult) -> dict[str, Any]:
         }
 
     USAGE_DISPLAY_LIMIT = 5
+    USAGE_MESSAGE_DISPLAY_LIMIT = 3
+
+    @staticmethod
+    def build_workflow_in_use_message(workflow_name: str, usage: dict[str, Any]) -> str:
+        """Builds a user-facing message listing pipelines/APIs blocking deletion.
+
+        Matches the format used by the frontend so direct API callers see the
+        same details about which pipelines/API deployments are using the WF.
+        """
+        pipelines = usage.get("pipelines") or []
+        api_names = usage.get("api_names") or []
+        pipeline_count = usage.get("pipeline_count", 0)
+        api_count = usage.get("api_count", 0)
+
+        if (pipeline_count + api_count) == 0:
+            return f"Cannot delete `{workflow_name}` as it is currently in use."
+
+        limit = WorkflowHelper.USAGE_MESSAGE_DISPLAY_LIMIT
+        lines: list[str] = []
+
+        if api_names:
+            shown = list(api_names)[:limit]
+            for name in shown:
+                lines.append(f"- `{name}` (API Deployment)")
+            remaining = api_count - len(shown)
+            if remaining > 0:
+                lines.append(f"- ...and {remaining} more API deployment(s)")
+
+        if pipelines:
+            shown = list(pipelines)[:limit]
+            for p in shown:
+                name = p.get("pipeline_name")
+                p_type = p.get("pipeline_type")
+                lines.append(f"- `{name}` ({p_type} Pipeline)")
+            remaining = pipeline_count - len(shown)
+            if remaining > 0:
+                lines.append(f"- ...and {remaining} more pipeline(s)")
+
+        details = "\n".join(lines)
+        return f"Cannot delete `{workflow_name}` as it is used in:\n{details}"
 
     @staticmethod
     def can_update_workflow(workflow_id: str) -> dict[str, Any]: