feat: add helpers for suspend and retry pipelines

kratix_sdk/kratix_sdk.py

@@ -174,3 +174,36 @@ def is_configure_action(self) -> bool:
     def is_delete_action(self) -> bool:
         """Returns true if the workflow is a delete action."""
         return self.workflow_action() == "delete"
+
+    def suspend(self, message: str = "") -> None:
+        """Suspends the pipeline by writing workflow-control.yaml with suspend: true.
+
+        Kratix will stop further pipeline execution and set the workflow phase to
+        Suspended.
+
+        If a message is provided, it will be surfaced in the object's status."""
+        data: dict = {"suspend": True}
+        if message:
+            data["message"] = message
+        self._write_workflow_control(data)
+
+    def retry_after(self, duration: str, message: str = "") -> None:
+        """Configures the pipeline to be retried after a given duration.
+
+        The duration must be a valid Go duration string (e.g. "5m", "1h30m", "300ms").
+        Kratix will requeue the pipeline after the specified duration and increment
+        the attempt counter in the object's status.
+
+        If a message is provided, it will be surfaced in the object's status."""
+        if not duration:
+            raise ValueError("duration must be a non-empty string")
+        data: dict = {"retryAfter": duration}
+        if message:
+            data["message"] = message
+        self._write_workflow_control(data)
+
+    def _write_workflow_control(self, data: dict) -> None:
+        path = METADATA_DIR / "workflow-control.yaml"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open("w") as f:
+            yaml.safe_dump(data, f)

tests/test_kratix_sdk.py

@@ -235,6 +235,59 @@ def test_is_delete_action(monkeypatch):
     assert sdk.is_delete_action() is False
 
 
+# ---------- Suspend / Retry Tests ----------
+
+
+def test_suspend_writes_workflow_control():
+    sdk = ks.KratixSDK()
+
+    sdk.suspend()
+
+    written = yaml.safe_load(
+        (ks.get_metadata_dir() / "workflow-control.yaml").read_text()
+    )
+    assert written == {"suspend": True}
+
+
+def test_suspend_with_message():
+    sdk = ks.KratixSDK()
+
+    sdk.suspend(message="waiting for dependency")
+
+    written = yaml.safe_load(
+        (ks.get_metadata_dir() / "workflow-control.yaml").read_text()
+    )
+    assert written == {"suspend": True, "message": "waiting for dependency"}
+
+
+def test_retry_after_writes_workflow_control():
+    sdk = ks.KratixSDK()
+
+    sdk.retry_after("5m")
+
+    written = yaml.safe_load(
+        (ks.get_metadata_dir() / "workflow-control.yaml").read_text()
+    )
+    assert written == {"retryAfter": "5m"}
+
+
+def test_retry_after_with_message():
+    sdk = ks.KratixSDK()
+
+    sdk.retry_after("1h30m", message="configmap not found yet")
+
+    written = yaml.safe_load(
+        (ks.get_metadata_dir() / "workflow-control.yaml").read_text()
+    )
+    assert written == {"retryAfter": "1h30m", "message": "configmap not found yet"}
+
+
+def test_retry_after_empty_duration_raises():
+    sdk = ks.KratixSDK()
+    with pytest.raises(ValueError):
+        sdk.retry_after("")
+
+
 # ---------- Write to Output Tests ----------