chore: workflow full sync (keephq#3383)

.github/workflows/auto-resolve-keep.yml

@@ -16,8 +16,9 @@ on:
         required: false
         type: string
         default: "resolved"
-  pull_request:
-    types: [opened, edited, closed]
+  push:
+    branches:
+      - main
 
 jobs:
   auto-resolve-keep:

examples/workflows/update_workflows_from_s3.yml

@@ -12,6 +12,13 @@ steps:
         bucket: "keep-workflows"
 
 actions:
+  # optional: delete all workflows before updating for full sync
+  # - name: full-sync
+  #   provider:
+  #     type: keep
+  #     with:
+  #       workflow_full_sync: true
+
   - name: update
     foreach: "{{ steps.s3-dump.results }}"
     provider:

keep/cli/cli.py

@@ -68,7 +68,7 @@ def get_default_conf_file_path():
 
 def make_keep_request(method, url, **kwargs):
     if os.environ.get("KEEP_CLI_IGNORE_SSL", "false").lower() == "true":
-        kwargs['verify'] = False
+        kwargs["verify"] = False
     try:
         response = requests.request(method, url, **kwargs)
         if response.status_code == 401:
@@ -404,6 +404,26 @@ def list_workflows(info: Info):
     print(table)
 
 
+def get_workflows(info: Info):
+    """Get all workflows."""
+    resp = make_keep_request(
+        "GET",
+        info.keep_api_url + "/workflows",
+        headers={"x-api-key": info.api_key, "accept": "application/json"},
+    )
+    return resp.json()
+
+
+def delete_workflow(workflow_id: str, info: Info):
+    """Delete a workflow."""
+    resp = make_keep_request(
+        "DELETE",
+        info.keep_api_url + f"/workflows/{workflow_id}",
+        headers={"x-api-key": info.api_key, "accept": "application/json"},
+    )
+    return resp
+
+
 def apply_workflow(file: str, info: Info):
     """Helper function to apply a single workflow."""
     with open(file, "rb") as f:
@@ -426,10 +446,36 @@ def apply_workflow(file: str, info: Info):
     help="The workflow file or directory containing workflow files",
     required=True,
 )
+@click.option(
+    "--full-sync",
+    is_flag=True,
+    help="Delete all existing workflows and apply the new ones",
+    default=False,
+)
 @pass_info
-def apply(info: Info, file: str):
+def apply(info: Info, file: str, full_sync: bool):
     """Apply a workflow or multiple workflows from a directory."""
     if os.path.isdir(file):
+        if full_sync:
+            click.echo(click.style("Deleting all workflows", bold=True))
+            workflows = get_workflows(info)
+            for workflow in workflows:
+                click.echo(
+                    click.style(f"Deleting workflow {workflow['id']}", bold=True)
+                )
+                resp = delete_workflow(workflow["id"], info)
+                if resp.ok:
+                    click.echo(
+                        click.style(f"Deleted workflow {workflow['id']}", bold=True)
+                    )
+                else:
+                    click.echo(
+                        click.style(
+                            f"Error deleting workflow {workflow['id']}: {resp.text}",
+                            bold=True,
+                        )
+                    )
+            click.echo(click.style("Deleted all workflows", bold=True))
         for filename in os.listdir(file):
             if filename.endswith(".yml") or filename.endswith(".yaml"):
                 click.echo(click.style(f"Applying workflow {filename}", bold=True))

keep/providers/keep_provider/keep_provider.py

@@ -542,8 +542,30 @@ def _notify_alert(self, **kwargs):
         )
         return alerts
 
+    def _delete_workflows(self):
+        self.logger.info("Deleting all workflows")
+        workflow_store = WorkflowStore()
+        workflows = workflow_store.get_all_workflows(self.context_manager.tenant_id)
+        for workflow in workflows:
+            self.logger.info(f"Deleting workflow {workflow['id']}")
+            try:
+                workflow_store.delete_workflow(
+                    self.context_manager.tenant_id, workflow.id
+                )
+                self.logger.info(f"Deleted workflow {workflow['id']}")
+            except Exception as e:
+                self.logger.exception(
+                    f"Failed to delete workflow {workflow['id']}: {e}"
+                )
+                raise ProviderException(
+                    f"Failed to delete workflow {workflow['id']}: {e}"
+                )
+        self.logger.info("Deleted all workflows")
+
     def _notify(self, **kwargs):
-        if "workflow_to_update_yaml" in kwargs:
+        if "workflow_full_sync" in kwargs:
+            self._delete_workflows()
+        elif "workflow_to_update_yaml" in kwargs:
             workflow_to_update_yaml = kwargs["workflow_to_update_yaml"]
             self.logger.info(
                 "Updating workflow YAML",