add purge api

Signed-off-by: Fabian Martinez <[email protected]>

Author: famarting

durabletask/client.py

@@ -199,3 +199,8 @@ def resume_orchestration(self, instance_id: str):
         req = pb.ResumeRequest(instanceId=instance_id)
         self._logger.info(f"Resuming instance '{instance_id}'.")
         self._stub.ResumeInstance(req)
+
+    def purge_orchestration(self, instance_id: str, recursive: bool = True):
+        req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
+        self._logger.info(f"Purging instance '{instance_id}'.")
+        self._stub.PurgeInstances()

tests/test_orchestration_e2e.py

@@ -309,6 +309,9 @@ def child(ctx: task.OrchestrationContext, _):
         assert state is not None
         assert state.runtime_status == client.OrchestrationStatus.TERMINATED
 
+        task_hub_client.purge_orchestration(id)
+        state = task_hub_client.get_orchestration_state(id)
+        assert state is None
 
 
 def test_continue_as_new():