Close existing PR on update

smkent · Feb 21, 2023 · e89cc06 · e89cc06
1 parent 0305500
commit e89cc06
Showing 1 changed file with 47 additions and 0 deletions.
diff --git a/cookie_python/manage/repo.py b/cookie_python/manage/repo.py
@@ -107,7 +107,54 @@ def lint_test(self) -> None:
             self.logger.error("Resolve errors and exit shell to continue")
             self.shell()
 
+    def find_existing_pr(self) -> Optional[str]:
+        with contextlib.suppress(
+            subprocess.CalledProcessError, json.JSONDecodeError, TypeError
+        ):
+            for pr in json.loads(
+                self.run(
+                    [
+                        "gh",
+                        "pr",
+                        "list",
+                        "-H",
+                        self.branch,
+                        "-B",
+                        "main",
+                        "--json",
+                        ",".join(("url", "headRefName", "baseRefName")),
+                    ],
+                    capture_output=True,
+                    check=True,
+                ).stdout.decode()
+            ):
+                pr_url = str(pr.pop("url"))
+                if pr == {"headRefName": self.branch, "baseRefName": "main"}:
+                    return pr_url
+        return None
+
+    def close_existing_pr(self) -> None:
+        # Locate existing PR
+        pr_url = self.find_existing_pr()
+        if pr_url:
+            if self.dry_run:
+                self.logger.info(f"Would close existing PR {pr_url}")
+            else:
+                self.run(["gh", "pr", "close", pr_url])
+                self.logger.info(f"Closed existing PR {pr_url}")
+        if self.dry_run:
+            return
+        # Delete existing branch
+        delete_result = self.run(
+            ["git", "push", "origin", f":{self.branch}"],
+            capture_output=True,
+            check=False,
+        )
+        if delete_result.returncode == 0:
+            self.logger.info(f"Deleted existing remote branch {self.branch}")
+
     def open_pr(self, message: str) -> None:
+        self.close_existing_pr()
         if self.dry_run:
             self.logger.success("Would open PR")
             return