Use PyGithub for pull request creation, deletion

Author: smkent

cookie_python/manage/github.py

@@ -1,8 +1,10 @@
 import contextlib
 import os
 from functools import cached_property, lru_cache
+from typing import Optional
 
 from github import Github
+from github.PullRequest import PullRequest
 from github.Repository import Repository
 
 
@@ -22,3 +24,16 @@ def find_repo(self, search: str) -> Repository:
             with contextlib.suppress(IndexError):
                 search = os.path.splitext(search.split(":")[1])[0]
         return self._gh.get_repo(search)
+
+    def find_pr(
+        self, repo: Repository, head: str, base: str = "main"
+    ) -> Optional[PullRequest]:
+        pulls = [
+            pr
+            for pr in repo.get_pulls(head=f"{self.username}:{head}", base=base)
+        ]
+        if not pulls:
+            return None
+        if len(pulls) > 1:
+            raise Exception(f"Multiple PRs found matching {head}")
+        return pulls[0]

cookie_python/manage/repo.py

@@ -115,41 +115,15 @@ 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:
+        pr = self.gh.find_pr(self.repo, self.branch)
+        if pr:
             if self.dry_run:
-                self.logger.info(f"Would close existing PR {pr_url}")
+                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}")
+                pr.edit(state="closed")
+                self.logger.info(f"Closed existing PR {pr.url}")
         if self.dry_run:
             return
         # Delete existing branch
@@ -168,21 +142,10 @@ def open_pr(self, message: str) -> None:
             return
         self.run(["git", "push", "origin", self.branch])
         commit_title, _, *commit_body = message.splitlines()
-        pr_url = self.run(
-            [
-                "gh",
-                "pr",
-                "create",
-                "--title",
-                commit_title.strip(),
-                "--body-file",
-                "-",
-                "--base",
-                "main",
-                "--head",
-                self.branch,
-            ],
-            input=os.linesep.join(commit_body).encode("utf-8"),
-            capture_output=True,
-        ).stdout.decode()
-        self.logger.success(f"Opened PR {pr_url}")
+        pr = self.repo.create_pull(
+            base="main",
+            head=self.branch,
+            title=commit_title.strip(),
+            body=os.linesep.join(commit_body),
+        )
+        self.logger.success(f"Opened PR {pr.url}")