feat(workflow-triggers): add pause and resume functionality

CHANGELOG.md

@@ -13,6 +13,17 @@ As of 2025-08-29, changes are grouped as follows
 - 🐛 Bug Fixes: Bug fixes.
 - ⚡ Improvements: Transparent changes, e.g. better performance.
 
+## [7.90.0] (2025-11-26)
+
+
+### Features
+
+* **workflows:** Add pause and resume functionality for workflow triggers ([ca58b39](https://github.com/cognitedata/cognite-sdk-python/commit/ca58b391)) 
+  - Add `pause()` and `resume()` methods to WorkflowTriggerAPI
+  - Add `is_paused` field to WorkflowTrigger data class
+  - Support for pausing and resuming scheduled and data modeling triggers
+  - Includes comprehensive unit and integration tests
+
 ## [7.89.0](https://github.com/cognitedata/cognite-sdk-python/compare/v7.88.0...v7.89.0) (2025-11-04)
 
 

cognite/client/_api/workflows/triggers.py

@@ -217,3 +217,47 @@ def list_runs(self, external_id: str, limit: int | None = DEFAULT_LIMIT_READ) ->
             list_cls=WorkflowTriggerRunList,
             limit=limit,
         )
+
+    def pause(self, external_id: str) -> None:
+        """`Pause a workflow trigger. <https://api-docs.cognite.com/20230101/tag/Workflow-triggers/operation/pauseTrigger>`_
+
+        When a trigger is paused, it will not trigger new workflow executions.
+        This operation is idempotent - pausing an already paused trigger has no effect.
+
+        Args:
+            external_id (str): The external id of the trigger to pause.
+
+        Examples:
+
+            Pause a trigger:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> client.workflows.triggers.pause("my_trigger")
+        """
+        self._post(
+            url_path=interpolate_and_url_encode(self._RESOURCE_PATH + "/{}/pause", external_id),
+            json={},
+        )
+
+    def resume(self, external_id: str) -> None:
+        """`Resume a paused workflow trigger. <https://api-docs.cognite.com/20230101/tag/Workflow-triggers/operation/resumeTrigger>`_
+
+        When a trigger is resumed, it will start triggering workflow executions again according to its schedule.
+        This operation is idempotent - resuming an already running trigger has no effect.
+
+        Args:
+            external_id (str): The external id of the trigger to resume.
+
+        Examples:
+
+            Resume a trigger:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> client.workflows.triggers.resume("my_trigger")
+        """
+        self._post(
+            url_path=interpolate_and_url_encode(self._RESOURCE_PATH + "/{}/resume", external_id),
+            json={},
+        )

cognite/client/data_classes/workflows.py

@@ -1583,6 +1583,7 @@ class WorkflowTrigger(WorkflowTriggerCore):
         metadata (dict | None): Application specific metadata. Defaults to None.
         created_time (int | None): The time when the workflow version trigger was created. Unix timestamp in milliseconds. Defaults to None.
         last_updated_time (int | None): The time when the workflow version trigger was last updated. Unix timestamp in milliseconds. Defaults to None.
+        is_paused (bool | None): Whether the trigger is paused. Defaults to None.
     """
 
     def __init__(
@@ -1595,6 +1596,7 @@ def __init__(
         metadata: dict | None = None,
         created_time: int | None = None,
         last_updated_time: int | None = None,
+        is_paused: bool | None = None,
     ) -> None:
         super().__init__(
             external_id=external_id,
@@ -1606,6 +1608,7 @@ def __init__(
         )
         self.created_time = created_time
         self.last_updated_time = last_updated_time
+        self.is_paused = is_paused
 
     def dump(self, camel_case: bool = True) -> dict[str, Any]:
         item: dict[str, Any] = {
@@ -1622,6 +1625,8 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
             item["created_time"] = self.created_time
         if self.last_updated_time:
             item["last_updated_time"] = self.last_updated_time
+        if self.is_paused is not None:
+            item["is_paused"] = self.is_paused
         if camel_case:
             return convert_all_keys_to_camel_case(item)
         return item
@@ -1637,6 +1642,7 @@ def _load(cls, resource: dict, cognite_client: CogniteClient | None = None) -> W
             metadata=resource.get("metadata"),
             created_time=resource.get("createdTime"),
             last_updated_time=resource.get("lastUpdatedTime"),
+            is_paused=resource.get("isPaused"),
         )
 
     def as_write(self) -> WorkflowTriggerUpsert:

docs/source/data_workflows.rst

@@ -87,6 +87,14 @@ Delete Triggers
 ^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.workflows.WorkflowTriggerAPI.delete
 
+Pause Trigger
+^^^^^^^^^^^^^
+.. automethod:: cognite.client._api.workflows.WorkflowTriggerAPI.pause
+
+Resume Trigger
+^^^^^^^^^^^^^^
+.. automethod:: cognite.client._api.workflows.WorkflowTriggerAPI.resume
+
 
 Data Workflows data classes
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/tests_integration/test_api/test_workflows.py

@@ -761,3 +761,71 @@ def test_trigger_run_history_non_existing(self, cognite_client: CogniteClient) -
             cognite_client.workflows.triggers.get_trigger_run_history(
                 external_id="integration_test-non_existing_trigger"
             )
+
+    def test_pause_resume_trigger(
+        self,
+        cognite_client: CogniteClient,
+        permanent_scheduled_trigger: WorkflowTrigger,
+    ) -> None:
+        # Get initial state
+        initial_trigger = cognite_client.workflows.triggers.list(limit=100)
+        trigger = next((t for t in initial_trigger if t.external_id == permanent_scheduled_trigger.external_id), None)
+        assert trigger is not None
+        initial_paused_state = trigger.is_paused
+
+        # Test pause operation
+        cognite_client.workflows.triggers.pause(permanent_scheduled_trigger.external_id)
+
+        # Verify trigger is paused
+        paused_triggers = cognite_client.workflows.triggers.list(limit=100)
+        paused_trigger = next(
+            (t for t in paused_triggers if t.external_id == permanent_scheduled_trigger.external_id), None
+        )
+        assert paused_trigger is not None
+        assert paused_trigger.is_paused is True
+
+        # Test idempotency - pause again
+        cognite_client.workflows.triggers.pause(permanent_scheduled_trigger.external_id)
+
+        # Verify still paused
+        still_paused_triggers = cognite_client.workflows.triggers.list(limit=100)
+        still_paused_trigger = next(
+            (t for t in still_paused_triggers if t.external_id == permanent_scheduled_trigger.external_id), None
+        )
+        assert still_paused_trigger is not None
+        assert still_paused_trigger.is_paused is True
+
+        # Test resume operation
+        cognite_client.workflows.triggers.resume(permanent_scheduled_trigger.external_id)
+
+        # Verify trigger is resumed
+        resumed_triggers = cognite_client.workflows.triggers.list(limit=100)
+        resumed_trigger = next(
+            (t for t in resumed_triggers if t.external_id == permanent_scheduled_trigger.external_id), None
+        )
+        assert resumed_trigger is not None
+        assert resumed_trigger.is_paused is False
+
+        # Test idempotency - resume again
+        cognite_client.workflows.triggers.resume(permanent_scheduled_trigger.external_id)
+
+        # Verify still resumed
+        still_resumed_triggers = cognite_client.workflows.triggers.list(limit=100)
+        still_resumed_trigger = next(
+            (t for t in still_resumed_triggers if t.external_id == permanent_scheduled_trigger.external_id), None
+        )
+        assert still_resumed_trigger is not None
+        assert still_resumed_trigger.is_paused is False
+
+        # Restore initial state if it was paused
+        if initial_paused_state:
+            cognite_client.workflows.triggers.pause(permanent_scheduled_trigger.external_id)
+
+    def test_pause_resume_nonexistent_trigger(self, cognite_client: CogniteClient) -> None:
+        # Test pause on non-existent trigger
+        with pytest.raises(CogniteAPIError, match=r"Trigger not found\."):
+            cognite_client.workflows.triggers.pause("integration_test-non_existing_trigger")
+
+        # Test resume on non-existent trigger
+        with pytest.raises(CogniteAPIError, match=r"Trigger not found\."):
+            cognite_client.workflows.triggers.resume("integration_test-non_existing_trigger")

tests/tests_unit/test_data_classes/test_workflows.py

@@ -18,8 +18,10 @@
     WorkflowDefinitionUpsert,
     WorkflowExecutionDetailed,
     WorkflowIds,
+    WorkflowScheduledTriggerRule,
     WorkflowTask,
     WorkflowTaskOutput,
+    WorkflowTrigger,
     WorkflowVersionId,
 )
 
@@ -248,3 +250,104 @@ class TestWorkflowTask:
     def test_serialization(self, raw: dict):
         loaded = WorkflowTask._load(raw)
         assert loaded.dump() == raw
+
+
+class TestWorkflowTrigger:
+    def test_load_with_is_paused_true(self):
+        """Test loading WorkflowTrigger from JSON with isPaused=true."""
+        raw_data = {
+            "externalId": "test-trigger",
+            "workflowExternalId": "test-workflow",
+            "workflowVersion": "1.0",
+            "triggerRule": {"triggerType": "scheduled", "cronExpression": "0 0 * * *"},
+            "createdTime": 1696240547972,
+            "lastUpdatedTime": 1696240548972,
+            "isPaused": True,
+        }
+
+        trigger = WorkflowTrigger._load(raw_data)
+
+        assert trigger.external_id == "test-trigger"
+        assert trigger.workflow_external_id == "test-workflow"
+        assert trigger.workflow_version == "1.0"
+        assert trigger.is_paused is True
+        assert trigger.created_time == 1696240547972
+        assert trigger.last_updated_time == 1696240548972
+
+    def test_load_with_is_paused_false(self):
+        """Test loading WorkflowTrigger from JSON with isPaused=false."""
+        raw_data = {
+            "externalId": "test-trigger",
+            "workflowExternalId": "test-workflow",
+            "workflowVersion": "1.0",
+            "triggerRule": {"triggerType": "scheduled", "cronExpression": "0 0 * * *"},
+            "isPaused": False,
+        }
+
+        trigger = WorkflowTrigger._load(raw_data)
+
+        assert trigger.external_id == "test-trigger"
+        assert trigger.is_paused is False
+
+    def test_load_without_is_paused(self):
+        """Test loading WorkflowTrigger from JSON without isPaused field."""
+        raw_data = {
+            "externalId": "test-trigger",
+            "workflowExternalId": "test-workflow",
+            "workflowVersion": "1.0",
+            "triggerRule": {"triggerType": "scheduled", "cronExpression": "0 0 * * *"},
+        }
+
+        trigger = WorkflowTrigger._load(raw_data)
+
+        assert trigger.external_id == "test-trigger"
+        assert trigger.is_paused is None
+
+    def test_dump_with_is_paused_true(self):
+        """Test dumping WorkflowTrigger to JSON with isPaused=true."""
+        trigger = WorkflowTrigger(
+            external_id="test-trigger",
+            workflow_external_id="test-workflow",
+            workflow_version="1.0",
+            trigger_rule=WorkflowScheduledTriggerRule(cron_expression="0 0 * * *"),
+            is_paused=True,
+        )
+
+        dumped = trigger.dump(camel_case=True)
+        assert dumped["isPaused"] is True
+
+        dumped_snake = trigger.dump(camel_case=False)
+        assert dumped_snake["is_paused"] is True
+
+    def test_dump_with_is_paused_false(self):
+        """Test dumping WorkflowTrigger to JSON with isPaused=false."""
+        trigger = WorkflowTrigger(
+            external_id="test-trigger",
+            workflow_external_id="test-workflow",
+            workflow_version="1.0",
+            trigger_rule=WorkflowScheduledTriggerRule(cron_expression="0 0 * * *"),
+            is_paused=False,
+        )
+
+        dumped = trigger.dump(camel_case=True)
+        assert dumped["isPaused"] is False
+
+        dumped_snake = trigger.dump(camel_case=False)
+        assert dumped_snake["is_paused"] is False
+
+    def test_dump_without_is_paused(self):
+        """Test dumping WorkflowTrigger to JSON without isPaused field (None)."""
+        trigger = WorkflowTrigger(
+            external_id="test-trigger",
+            workflow_external_id="test-workflow",
+            workflow_version="1.0",
+            trigger_rule=WorkflowScheduledTriggerRule(cron_expression="0 0 * * *"),
+            is_paused=None,
+        )
+
+        dumped = trigger.dump(camel_case=True)
+        # When None, the field should not be included in dump
+        assert "isPaused" not in dumped
+
+        dumped_snake = trigger.dump(camel_case=False)
+        assert "is_paused" not in dumped_snake