Reimagine auto-merge scripts as python

.github/workflows/auto-merge-successful-prs.yaml

@@ -3,14 +3,14 @@ name: Auto-merge Successful PRs
 on:
   workflow_dispatch:
   schedule:
-    - cron: "*/15 * * * *"  # Every 15 minutes
+    - cron: "0 */4 * * *"  # Every 4 hours
 
 permissions:
   contents: read
   pull-requests: write
 
 jobs:
-  find_and_merge:
+  merge-successful-prs:
     runs-on: ubuntu-latest
 
     steps:
@@ -22,8 +22,9 @@ jobs:
         uses: actions/checkout@v4
         with:
           ssh-key: ${{ secrets.BOT_SSH_KEY }}
-
-        # Fetch open pull requests and check for status checks on automerge PRs
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
       - name: Auto-merge pull requests if all status checks pass
         run: |
-            build-scripts/hack/auto-merge-successful-prs.sh
+            build-scripts/hack/auto-merge-successful-prs.py

.github/workflows/update-components.yaml

@@ -57,7 +57,5 @@ jobs:
           branch: "autoupdate/sync/${{ matrix.branch }}"
           labels: |
             automerge
-            automated pr
-            component update
           delete-branch: true
           base: ${{ matrix.branch }}

build-scripts/hack/auto-merge-successful-pr.py

@@ -0,0 +1,55 @@
+#!/bin/env python3
+
+import shlex
+import subprocess
+import json
+
+LABEL = "automerge"
+APPROVE_MSG = "All status checks passed for PR #{}."
+
+
+def sh(cmd: str) -> str:
+    """Run a shell command and return its output."""
+    _pipe = subprocess.PIPE
+    result = subprocess.run(shlex.split(cmd), stdout=_pipe, stderr=_pipe, text=True)
+    if result.returncode != 0:
+        raise Exception(f"Error running command: {cmd}\nError: {result.stderr}")
+    return result.stdout.strip()
+
+
+def get_pull_requests() -> list:
+    """Fetch open pull requests matching some label."""
+    prs_json = sh("gh pr list --state open --json number,labels")
+    prs = json.loads(prs_json)
+    return [pr for pr in prs if any(label["name"] == LABEL for label in pr["labels"])]
+
+
+def check_pr_passed(pr_number) -> bool:
+    """Check if all status checks passed for the given PR."""
+    checks_json = sh(f"gh pr checks {pr_number} --json bucket")
+    checks = json.loads(checks_json)
+    return all(check["bucket"] == "pass" for check in checks)
+
+
+def approve_and_merge_pr(pr_number) -> None:
+    """Approve and merge the PR."""
+    print(APPROVE_MSG.format(pr_number) + "Proceeding with merge...")
+    sh(f'gh pr review {pr_number} --approve -b "{APPROVE_MSG.format(pr_number)}"')
+    sh(f"gh pr merge {pr_number} --auto --squash")
+
+
+def process_pull_requests():
+    """Process the PRs and merge if checks have passed."""
+    prs = get_pull_requests()
+
+    for pr in prs:
+        pr_number: int = pr["number"]
+
+        if check_pr_passed(pr_number):
+            approve_and_merge_pr(pr_number)
+        else:
+            print(f"Status checks have not passed for PR #{pr_number}. Skipping merge.")
+
+
+if __name__ == "__main__":
+    process_pull_requests()