Migrate from github3.py to PyGithub

Author: RadekManak

pyproject.toml

@@ -10,7 +10,7 @@ readme = "README.md"
 requires-python = ">=3.11"
 
 dependencies = [
-    "github3-py==4.0.1",
+    "PyGithub==2.9.1",
     "gitpython==3.1.46",
     "requests==2.33.1",
 ]

rebasebot/bot.py

@@ -25,12 +25,11 @@
 
 import git
 import git.compat
-import github3
 import requests
 from git.objects import Commit
-from github3.pulls import ShortPullRequest
-from github3.repos.commit import ShortCommit
-from github3.repos.repo import Repository
+from github import GithubException
+from github.PullRequest import PullRequest
+from github.Repository import Repository
 
 from rebasebot import lifecycle_hooks
 from rebasebot.github import GithubAppProvider, GitHubBranch
@@ -82,9 +81,9 @@ def _needs_rebase(gitwd: git.Repo, source: GitHubBranch, dest: GitHubBranch) ->
 
 def _is_pr_merged(pr_number: int, source_repo: Repository, gitwd: git.Repo, source_branch: str) -> bool:
     logging.info("Checking that PR %s has been merged and is included in %s", pr_number, source_branch)
-    gh_pr = source_repo.pull_request(pr_number)
+    gh_pr = source_repo.get_pull(pr_number)
 
-    if not gh_pr.is_merged():
+    if not gh_pr.merged:
         return False
 
     merge_commit_sha = gh_pr.merge_commit_sha
@@ -633,11 +632,10 @@ def _cherrypick_art_pull_request(
     that updates the build image, it includes it in the rebase.
     """
     logging.info("Checking for ART pull request")
-    for pull_request in dest_repo.pull_requests(state="open", base=f"{dest.branch}"):
-        assert isinstance(pull_request, ShortPullRequest)  # type hint
+    for pull_request in dest_repo.get_pulls(state="open", base=dest.branch):
         if "consistent with ART" in pull_request.title and pull_request.user.login == "openshift-bot":
             logging.info(f"Found open ART image update pull requst: {pull_request.title}")
-            remote = pull_request.head.repository
+            remote = pull_request.head.repo
             remote_name = remote.name
             if remote_name in gitwd.remotes:
                 gitwd.remotes[remote_name].set_url(remote.html_url)
@@ -646,8 +644,7 @@ def _cherrypick_art_pull_request(
 
             gitwd.remotes[remote_name].fetch(pull_request.head.ref)
 
-            for commit in pull_request.commits():
-                assert isinstance(commit, ShortCommit)
+            for commit in pull_request.get_commits():
                 _safe_cherry_pick(
                     gitwd=gitwd,
                     sha=commit.sha,
@@ -684,14 +681,19 @@ def _is_pr_required(gitwd: git.Repo, rebase: GitHubBranch, dest: GitHubBranch) -
     return True
 
 
-def _is_pr_available(dest_repo: Repository, dest: GitHubBranch, rebase: GitHubBranch) -> tuple[ShortPullRequest, bool]:
+def _is_pr_available(
+    dest_repo: Repository, dest: GitHubBranch, rebase: GitHubBranch
+) -> tuple[PullRequest | None, bool]:
     logging.info("Checking for existing pull request")
 
-    pull_requests = dest_repo.pull_requests(base=dest.branch, state="open")
+    pull_requests = dest_repo.get_pulls(base=dest.branch, state="open")
     # Github does not support filtering cross-repository pull requests if both repositories
     # are owned by the same organization. We must filter client side.
     for pr in pull_requests:
-        pr_repo = pr.as_dict()["head"]["repo"]["full_name"]
+        if pr.head.repo is None:
+            logging.warning(f"Skipping PR with deleted head repository: {pr.html_url}")
+            continue
+        pr_repo = pr.head.repo.full_name
         if pr_repo == f"{rebase.ns}/{rebase.name}" and pr.head.ref == rebase.branch:
             logging.info('Found existing pull request: "%s" %s', pr.title, pr.html_url)
             return pr, True
@@ -702,7 +704,7 @@ def _is_pr_available(dest_repo: Repository, dest: GitHubBranch, rebase: GitHubBr
 
 def _create_pr(
     *,
-    gh_app: github3.GitHub,
+    gh_app,
     dest: GitHubBranch,
     source: GitHubBranch,
     rebase: GitHubBranch,
@@ -717,30 +719,19 @@ def _create_pr(
 
     logging.info("Creating a pull request")
 
-    # FIXME(rmanak): This hack is because github3 doesn't support setting
-    # head_repo param when creating a PR.
-    #
-    # This param is required when creating cross-repository pull requests if both repositories
-    # are owned by the same organization.
-    #
-    # https://github.com/sigmavirus24/github3.py/issues/1190
-
-    gh_pr: requests.Response = gh_app._post(  # pylint: disable=W0212
-        f"https://api.github.com/repos/{dest.ns}/{dest.name}/pulls",
-        data={
-            "title": title,
-            "head": rebase.branch,
-            "head_repo": f"{rebase.ns}/{rebase.name}",
-            "base": dest.branch,
-            "maintainer_can_modify": False,
-        },
-        json=True,
+    dest_repo = gh_app.get_repo(f"{dest.ns}/{dest.name}")
+
+    pr = dest_repo.create_pull(
+        title=title,
+        body="",
+        head=f"{rebase.ns}:{rebase.branch}",
+        base=dest.branch,
+        maintainer_can_modify=False,
     )
 
-    logging.debug(gh_pr.json())
-    gh_pr.raise_for_status()
+    logging.info(f"Created pull request: {pr.html_url}")
 
-    return gh_pr.json()["html_url"]
+    return pr.html_url
 
 
 def is_ref_a_tag(gitwd: git.Repo, ref: str) -> bool:
@@ -867,12 +858,12 @@ def _init_working_dir(
     return gitwd
 
 
-def _manual_rebase_pr_in_repo(repo: Repository) -> ShortPullRequest | None:
+def _manual_rebase_pr_in_repo(repo: Repository) -> PullRequest | None:
     """Checks for the presence of a rebase/manual label on the pull request."""
-    prs = repo.pull_requests()
+    prs = repo.get_pulls()
     for pull_req in prs:
         for label in pull_req.labels:
-            if label["name"] == "rebase/manual":
+            if label.name == "rebase/manual":
                 return pull_req
     return None
 
@@ -885,7 +876,7 @@ def _push_rebase_branch(gitwd: git.Repo, rebase: GitHubBranch) -> None:
         raise builtins.Exception(f"Error pushing to {rebase}: {result[0].summary}")
 
 
-def _update_pr_title(gitwd: git.Repo, pull_req: ShortPullRequest, source: GitHubBranch, dest: GitHubBranch) -> None:
+def _update_pr_title(gitwd: git.Repo, pull_req: PullRequest, source: GitHubBranch, dest: GitHubBranch) -> None:
     """Updates the pull request title to match the current state of the rebase branch
     Only updates the title if the title contains the word Merge.
     Keeping everything before "Merge" and updating everything after.
@@ -902,8 +893,10 @@ def _update_pr_title(gitwd: git.Repo, pull_req: ShortPullRequest, source: GitHub
             return
 
         logging.info(f"Updating pull request title: {computed_title}")
-        if not pull_req.update(title=computed_title):
-            raise PullRequestUpdateException(f"Error updating title for pull request: {pull_req.html_url}")
+        try:
+            pull_req.edit(title=computed_title)
+        except Exception as ex:
+            raise PullRequestUpdateException(f"Error updating title for pull request: {pull_req.html_url}") from ex
     else:
         logging.info(
             f'Open pull request title "{pull_req.title}" does not match rebasebot format. Keeping the current title.'
@@ -979,11 +972,11 @@ def run(
         hooks = lifecycle_hooks.LifecycleHooks(tmp_script_dir=None, args=None)
 
     try:
-        dest_repo = gh_app.repository(dest.ns, dest.name)
+        dest_repo = gh_app.get_repo(f"{dest.ns}/{dest.name}")
         logging.info("Destination repository is %s", dest_repo.clone_url)
-        rebase_repo = gh_cloner_app.repository(rebase.ns, rebase.name)
+        rebase_repo = gh_cloner_app.get_repo(f"{rebase.ns}/{rebase.name}")
         logging.info("rebase repository is %s", rebase_repo.clone_url)
-        source_repo = gh_app.repository(source.ns, source.name)
+        source_repo = gh_app.get_repo(f"{source.ns}/{source.name}")
         logging.info("source repository is %s", source_repo.clone_url)
 
         if not ignore_manual_label:
@@ -1171,9 +1164,11 @@ def run(
             f"{ex}",
         )
         return False
-    except requests.exceptions.HTTPError as ex:
-        logging.error(f"Failed to create a pull request: {ex}\n Response: %s", ex.response.text)
-        _message_slack(slack_webhook, f"Failed to create a pull request: {ex}\n Response: {ex.response.text}")
+    except GithubException as ex:
+        logging.error(f"Failed to create a pull request: {ex}")
+        if ex.data:
+            logging.error(f"Response data: {ex.data}")
+        _message_slack(slack_webhook, f"Failed to create a pull request: {ex}")
 
         return False
     except Exception as ex:

rebasebot/cli.py

@@ -383,8 +383,8 @@ def main():
     """Rebase Bot entry point function."""
     args = _parse_cli_arguments()
 
-    # Silence info logs from github3
-    logger = logging.getLogger("github3")
+    # Silence info logs from PyGithub
+    logger = logging.getLogger("github")
     logger.setLevel(logging.WARN)
 
     slack_webhook = None

rebasebot/github.py

@@ -13,14 +13,13 @@
 #    under the License.
 """Contains GitHub related helper classes."""
 
-import builtins
 import logging
 import re
 from dataclasses import dataclass
 from functools import cached_property
 from urllib.parse import urlparse
 
-import github3
+from github import Auth, Github, GithubIntegration, UnknownObjectException
 
 logger = logging.getLogger()
 
@@ -119,6 +118,9 @@ def __init__(
         self._app_credentials = None
         self._cloner_app_credentials = None
 
+        if self.user_auth and not self.user_token:
+            raise ValueError("User authentication requires a GitHub user token")
+
         if not user_auth:
             if not all((app_id, app_key, dest_branch, cloner_id, cloner_key, rebase_branch)):
                 raise ValueError("Credentials for both, cloning and pushing app should be provided")
@@ -135,67 +137,122 @@ def get_app_token(self) -> str:
 
         :return: str
         """
-        return self.github_app.session.auth.token
+        if self.user_auth:
+            return self._require_user_token()
+
+        return self._get_installation_token(self._app_installation_context, "app")
 
     def get_cloner_token(self) -> str:
         """
         Get cloner app auth token
 
         :return: str
         """
-        return self.github_cloner_app.session.auth.token
+        if self.user_auth:
+            return self._require_user_token()
+
+        return self._get_installation_token(self._cloner_app_installation_context, "cloner")
 
     @cached_property
-    def github_app(self) -> github3.GitHub:
+    def github_app(self) -> Github:
         """
         Authenticated GitHub app.
 
         In case `user_auth` = True, returns app authenticated with user token.
         In app mode `app_id`, `app_key` and `dest_branch` will be used for app authentication.
 
-        :return: github3.GitHub
+        :return: Github
         """
         if self.user_auth:
             return self._get_github_user_logged_in_app()
 
-        return self._github_login_app(self._app_credentials)
+        return self._get_github_installation_app(self._app_installation_context)
 
     @cached_property
-    def github_cloner_app(self) -> github3.GitHub:
+    def github_cloner_app(self) -> Github:
         """
         Authenticated GitHub app.
 
         In case `user_auth` = True, returns app authenticated with user token.
         In app mode `cloner_id`, `cloner_key` and `rebase_branch`will be used for app authentication.
 
-        :return: github3.GitHub
+        :return: Github
         """
         if self.user_auth:
             return self._get_github_user_logged_in_app()
 
-        return self._github_login_app(self._cloner_app_credentials)
+        return self._get_github_installation_app(self._cloner_app_installation_context)
+
+    @cached_property
+    def _app_installation_context(self) -> tuple[GithubIntegration, int]:
+        return self._get_installation_context(self._require_credentials(self._app_credentials, "app"))
+
+    @cached_property
+    def _cloner_app_installation_context(self) -> tuple[GithubIntegration, int]:
+        return self._get_installation_context(self._require_credentials(self._cloner_app_credentials, "cloner"))
+
+    def _require_user_token(self) -> str:
+        if self.user_token is None:
+            raise RuntimeError("GitHub user token is unavailable")
+        return self.user_token
 
     @staticmethod
-    def _github_login_app(credentials: GitHubAppCredentials) -> github3.GitHub:
+    def _require_credentials(credentials: GitHubAppCredentials | None, role: str) -> GitHubAppCredentials:
+        if credentials is None:
+            raise RuntimeError(f"GitHub {role} credentials are unavailable")
+        return credentials
+
+    @staticmethod
+    def _get_github_installation_app(installation_context: tuple[GithubIntegration, int]) -> Github:
+        """
+        Build a Github client authenticated as a GitHub App installation.
+
+        :return: Github
+        """
+        integration, installation_id = installation_context
+        return integration.get_github_for_installation(installation_id)
+
+    @staticmethod
+    def _get_installation_token(installation_context: tuple[GithubIntegration, int], role: str) -> str:
+        """
+        Fetch a fresh access token for a GitHub App installation.
+
+        :return: access token
+        """
+        integration, installation_id = installation_context
+        token = integration.get_access_token(installation_id).token
+        if not token:
+            raise RuntimeError(f"GitHub {role} token is unavailable")
+        return token
+
+    @staticmethod
+    def _get_installation_context(credentials: GitHubAppCredentials) -> tuple[GithubIntegration, int]:
+        """
+        Authenticate as a GitHub App and locate its installation for a repository.
+
+        :return: tuple of (GithubIntegration, installation_id)
+        """
         logging.info("Logging to GitHub as an Application for repository %s", credentials.github_branch.url)
-        gh_app = github3.GitHub()
-        gh_app.login_as_app(credentials.app_key, credentials.app_id, expire_in=300)
         gh_branch = credentials.github_branch
 
+        # Create app authentication
+        app_auth = Auth.AppAuth(credentials.app_id, credentials.app_key.decode("utf-8"))
+        gi = GithubIntegration(auth=app_auth)
+
         try:
-            install = gh_app.app_installation_for_repository(owner=gh_branch.ns, repository=gh_branch.name)
-        except github3.exceptions.NotFoundError as err:
+            # Get installation for the repository
+            installation = gi.get_repo_installation(gh_branch.ns, gh_branch.name)
+        except UnknownObjectException as err:
             msg = (
                 f"App has not been authorized by {gh_branch.ns}, or repo {gh_branch.ns}/{gh_branch.name} does not exist"
             )
             logging.error(msg)
-            raise builtins.Exception(msg) from err
+            raise RuntimeError(msg) from err
 
-        gh_app.login_as_app_installation(credentials.app_key, credentials.app_id, install.id)
-        return gh_app
+        return gi, installation.id
 
-    def _get_github_user_logged_in_app(self) -> github3.GitHub:
+    def _get_github_user_logged_in_app(self) -> Github:
         logging.info("Logging to GitHub as a User")
-        gh_app = github3.GitHub()
-        gh_app.login(token=self.user_token)
+        auth = Auth.Token(self._require_user_token())
+        gh_app = Github(auth=auth)
         return gh_app