Merge pull request #118 from github-community-projects/migrate-to-pygithub

refactor: migrate from github3-py to PyGithub

Author: jmeridth

.github/linters/.mypy.ini

@@ -1,5 +1,5 @@
 [mypy]
 disable_error_code = attr-defined, import-not-found
 
-[mypy-github3.*]
+[mypy-github.*]
 ignore_missing_imports = True

auth.py

@@ -1,39 +1,6 @@
 """This is the module that contains functions related to authenticating to GitHub with a personal access token."""
 
-import github3
-import requests
-from requests.adapters import HTTPAdapter
-from urllib3.util.retry import Retry
-
-# Retry strategy: 5 retries with exponential backoff for transient errors
-RETRY_STRATEGY = Retry(
-    total=5,
-    backoff_factor=1,  # 1s, 2s, 4s, 8s, 16s
-    status_forcelist=[429, 500, 502, 503, 504],
-    raise_on_status=False,
-)
-REQUEST_TIMEOUT = 30  # seconds
-
-
-def _configure_session(github_connection: github3.GitHub) -> None:
-    """Mount retry adapter and set timeout on the github3 session."""
-    session = getattr(github_connection, "session", None)
-    if session is None:
-        return
-    adapter = HTTPAdapter(max_retries=RETRY_STRATEGY)
-    session.mount("https://", adapter)
-    session.mount("http://", adapter)
-    session.request = _timeout_wrapper(session.request, REQUEST_TIMEOUT)  # type: ignore[method-assign]
-
-
-def _timeout_wrapper(original_request, default_timeout):
-    """Wrap a session.request method to inject a default timeout."""
-
-    def wrapper(*args, **kwargs):
-        kwargs.setdefault("timeout", default_timeout)
-        return original_request(*args, **kwargs)
-
-    return wrapper
+from github import Auth, Github, GithubIntegration
 
 
 def auth_to_github(
@@ -43,7 +10,7 @@ def auth_to_github(
     gh_app_private_key_bytes: bytes,
     ghe: str,
     gh_app_enterprise_only: bool,
-) -> github3.GitHub:
+) -> Github:
     """
     Connect to GitHub.com or GitHub Enterprise, depending on env variables.
 
@@ -56,30 +23,34 @@ def auth_to_github(
         gh_app_enterprise_only (bool): Set this to true if the GH APP is created on GHE and needs to communicate with GHE api only
 
     Returns:
-        github3.GitHub: the GitHub connection object
+        Github: the GitHub connection object
     """
+    ghe = ghe.rstrip("/")
+
     if gh_app_id and gh_app_private_key_bytes and gh_app_installation_id:
+        app_auth = Auth.AppAuth(int(gh_app_id), gh_app_private_key_bytes.decode())
+        installation_auth = app_auth.get_installation_auth(int(gh_app_installation_id))
         if ghe and gh_app_enterprise_only:
-            gh = github3.github.GitHubEnterprise(url=ghe)
+            github_connection = Github(
+                base_url=f"{ghe}/api/v3",
+                auth=installation_auth,
+            )
         else:
-            gh = github3.github.GitHub()
-        gh.login_as_app_installation(
-            gh_app_private_key_bytes, str(gh_app_id), gh_app_installation_id
-        )
-        github_connection = gh
+            github_connection = Github(auth=installation_auth)
     elif ghe and token:
-        github_connection = github3.github.GitHubEnterprise(url=ghe, token=token)
+        github_connection = Github(
+            base_url=f"{ghe}/api/v3",
+            auth=Auth.Token(token),
+        )
     elif token:
-        github_connection = github3.login(token=token)
+        github_connection = Github(auth=Auth.Token(token))
     else:
         raise ValueError(
-            "GH_TOKEN or the set of [GH_APP_ID, GH_APP_INSTALLATION_ID, GH_APP_PRIVATE_KEY] environment variables are not set"
+            "GH_TOKEN or the set of [GH_APP_ID, GH_APP_INSTALLATION_ID, "
+            "GH_APP_PRIVATE_KEY] environment variables are not set"
         )
 
-    if not github_connection:
-        raise ValueError("Unable to authenticate to GitHub")
-    _configure_session(github_connection)
-    return github_connection  # type: ignore
+    return github_connection
 
 
 def get_github_app_installation_token(
@@ -99,31 +70,32 @@ def get_github_app_installation_token(
         gh_app_installation_id (str): the GitHub App Installation ID
 
     Returns:
-        str: the GitHub App token
+        str | None: the GitHub App token, or None if the request fails
     """
-    jwt_headers = github3.apps.create_jwt_headers(gh_app_private_key_bytes, gh_app_id)
-    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
-    url = f"{api_endpoint}/app/installations/{gh_app_installation_id}/access_tokens"
-
     try:
-        response = requests.post(url, headers=jwt_headers, json=None, timeout=5)
-        response.raise_for_status()
-    except requests.exceptions.RequestException as e:
+        ghe = ghe.rstrip("/")
+        app_auth = Auth.AppAuth(int(gh_app_id), gh_app_private_key_bytes.decode())
+        if ghe:
+            gi = GithubIntegration(auth=app_auth, base_url=f"{ghe}/api/v3")
+        else:
+            gi = GithubIntegration(auth=app_auth)
+        installation_token = gi.get_access_token(int(gh_app_installation_id))
+        return installation_token.token
+    except Exception as e:  # pylint: disable=broad-exception-caught
         print(f"Request failed: {e}")
         return None
-    return response.json().get("token")
 
 
 def get_team_members(
-    github_connection: github3.GitHub,
+    github_connection: Github,
     org: str,
     team_slug: str,
 ) -> list[str]:
     """
     Fetch the members of a GitHub team by slug.
 
     Args:
-        github_connection: Authenticated github3 connection.
+        github_connection: Authenticated GitHub connection.
         org: The organization that owns the team.
         team_slug: The team slug (e.g., "nux-reviewers").
 
@@ -132,21 +104,20 @@ def get_team_members(
         Returns an empty list if the team is not found or an error occurs.
     """
     try:
-        organization = github_connection.organization(org)
+        organization = github_connection.get_organization(org)
         if not organization:
             print(f"  ⚠️  Organization '{org}' not found, skipping team '{team_slug}'")
             return []
 
-        # team_by_name accepts a slug despite its name (hits /orgs/{org}/teams/{slug})
-        team = organization.team_by_name(team_slug)
+        team = organization.get_team_by_slug(team_slug)
         if not team:
             print(
                 f"  ⚠️  Team '{team_slug}' not found in '{org}', "
                 "skipping (check token permissions: read:org)"
             )
             return []
 
-        members = [m.login for m in team.members()]
+        members = [m.login for m in team.get_members()]
         print(f"  Resolved team {org}/{team_slug}: {len(members)} member(s)")
         return members
     except Exception as e:  # pylint: disable=broad-except

conflict_detector.py

@@ -156,9 +156,9 @@ def verify_conflict(
     when VERIFY_CONFLICTS=true.
     """
     try:
-        repo = github_connection.repository(owner, repo_name)  # type: ignore[union-attr]
-        pr_a = repo.pull_request(conflict.pr_a.number)
-        pr_b = repo.pull_request(conflict.pr_b.number)
+        repo = github_connection.get_repo(f"{owner}/{repo_name}")  # type: ignore[union-attr]
+        pr_a = repo.get_pull(conflict.pr_a.number)
+        pr_b = repo.get_pull(conflict.pr_b.number)
 
         # If either PR is not mergeable on its own, they likely conflict
         if pr_a.mergeable is False or pr_b.mergeable is False:

issue_writer.py

@@ -8,7 +8,7 @@
 from conflict_detector import ConflictCluster, cluster_conflicts
 
 if TYPE_CHECKING:
-    from github3.repos.repo import Repository
+    from github.Repository import Repository
 
 logger = logging.getLogger(__name__)
 
@@ -28,7 +28,7 @@ def create_or_update_issue(
     """Create or update an issue in the repository with conflict information.
 
     Args:
-        repo: A github3.py repository object.
+        repo: A PyGithub repository object.
         conflicts: List of ConflictResult objects for this repository.
         report_title: Title for the issue.
         dry_run: If True, log what would happen but make no API calls.
@@ -67,7 +67,7 @@ def create_or_update_issue(
 
 def _find_existing_issue(repo: Repository, title: str):
     """Search open issues for one matching the given title and hidden tag."""
-    for issue in repo.issues(state="open"):
+    for issue in repo.get_issues(state="open"):
         if issue.title == title and ISSUE_TAG in (issue.body or ""):
             return issue
     return None

pr_comment.py

@@ -70,8 +70,7 @@ def post_pr_comments(
 
     for repo_name in all_repo_names:
         conflicts = all_conflicts_by_repo.get(repo_name, [])
-        owner, repo = repo_name.split("/")
-        repo_obj = github_connection.repository(owner, repo)
+        repo_obj = github_connection.get_repo(repo_name)
 
         # Group active conflicts by PR
         pr_conflicts = group_conflicts_by_pr(conflicts) if conflicts else {}
@@ -163,8 +162,8 @@ def _find_existing_comments(repo: Any, pr_number: int) -> list[Any]:
         List of comment objects with the bot signature (may be empty).
     """
     try:
-        pr = repo.pull_request(pr_number)
-        return [c for c in pr.issue_comments() if COMMENT_SIGNATURE in c.body]
+        pr = repo.get_pull(pr_number)
+        return [c for c in pr.get_issue_comments() if COMMENT_SIGNATURE in c.body]
     except Exception as e:  # pylint: disable=broad-except
         logger.warning("Failed to check existing comments on PR #%s: %s", pr_number, e)
         return []
@@ -222,8 +221,8 @@ def _post_comment(repo, pr_number: int, body: str) -> bool:
         True if successful, False otherwise
     """
     try:
-        pr = repo.pull_request(pr_number)
-        pr.create_comment(body)
+        pr = repo.get_pull(pr_number)
+        pr.create_issue_comment(body=body)
         logger.info("Posted comment to PR #%s", pr_number)
         return True
     except Exception as e:  # pylint: disable=broad-except

pr_conflict_detector.py

@@ -217,8 +217,7 @@ def main():
         print("Report issue creation disabled (ENABLE_REPORT_ISSUES=false)")
     elif not env_vars.dry_run:
         for repo_full_name, conflicts in all_conflicts.items():
-            owner, rname = repo_full_name.split("/")
-            repo_obj = github_connection.repository(owner, rname)
+            repo_obj = github_connection.get_repo(repo_full_name)
             issue_url = create_or_update_issue(
                 repo_obj, conflicts, env_vars.report_title, env_vars.dry_run
             )
@@ -273,19 +272,18 @@ def get_repos_iterator(github_connection, env_vars):
     """Get an iterator of repositories to scan.
 
     Args:
-        github_connection: Authenticated github3 connection.
+        github_connection: Authenticated PyGithub connection.
         env_vars: Environment variables dataclass.
 
     Returns:
-        Iterator of github3 repository objects.
+        Iterator of PyGithub repository objects.
     """
     if env_vars.organization and not env_vars.repository_list:
-        return github_connection.organization(env_vars.organization).repositories()
+        return github_connection.get_organization(env_vars.organization).get_repos()
 
     repos = []
     for repo_full_name in env_vars.repository_list:
-        owner, repo_name = repo_full_name.split("/")
-        repos.append(github_connection.repository(owner, repo_name))
+        repos.append(github_connection.get_repo(repo_full_name))
     return repos
 
 

pr_data.py

@@ -59,17 +59,17 @@ def parse_patch_line_ranges(patch: str | None) -> list[tuple[int, int]]:
 
 def get_open_prs(repo: object, include_drafts: bool = True) -> list[PullRequestData]:
     """
-    Fetch all open PRs from a github3-py repository object.
+    Fetch all open PRs from a PyGithub repository object.
 
     Args:
-        repo: A github3-py repository object.
+        repo: A PyGithub repository object.
         include_drafts: If False, filter out draft PRs.
 
     Returns:
         A list of PullRequestData objects (without changed files populated).
     """
     prs: list[PullRequestData] = []
-    for pr in repo.pull_requests(state="open"):  # type: ignore[attr-defined]
+    for pr in repo.get_pulls(state="open"):  # type: ignore[attr-defined]
         is_draft = getattr(pr, "draft", False) or False
         if not include_drafts and is_draft:
             continue
@@ -97,12 +97,12 @@ def get_pr_changed_files(  # pylint: disable=unused-argument
     """
     Fetch the list of changed files for a given pull request.
 
-    Uses the github3-py pull request's files() method, then parses
+    Uses the PyGithub pull request's get_files() method, then parses
     each file's patch to extract modified line ranges.
 
     Args:
-        pull_request: A github3-py ShortPullRequest or PullRequest object.
-        github_connection: The github3-py GitHub connection (unused but kept
+        pull_request: A PyGithub ShortPullRequest or PullRequest object.
+        github_connection: The PyGithub GitHub connection (unused but kept
             for API consistency with other OSPO actions).
         owner: The repository owner.
         repo_name: The repository name.
@@ -111,7 +111,7 @@ def get_pr_changed_files(  # pylint: disable=unused-argument
         A list of ChangedFile objects.
     """
     changed_files: list[ChangedFile] = []
-    for f in pull_request.files():  # type: ignore[attr-defined]
+    for f in pull_request.get_files():  # type: ignore[attr-defined]
         patch = getattr(f, "patch", None)
         changed_files.append(
             ChangedFile(
@@ -137,9 +137,9 @@ def fetch_all_pr_data(
     Fetch all open PRs and their changed files from a repository.
 
     Args:
-        repo: A github3-py repository object.
+        repo: A PyGithub repository object.
         include_drafts: If False, filter out draft PRs.
-        github_connection: The github3-py GitHub connection.
+        github_connection: The PyGithub GitHub connection.
         owner: The repository owner.
         repo_name: The repository name.
         filter_authors: If provided, only fetch file changes for PRs authored
@@ -171,8 +171,8 @@ def fetch_all_pr_data(
             print(f"  Progress: {i + 1}/{total} PRs processed")
 
         try:
-            # Re-fetch the full PR object to call files()
-            full_pr = repo.pull_request(pr_data.number)  # type: ignore[attr-defined]
+            # Re-fetch the full PR object to call get_files()
+            full_pr = repo.get_pull(pr_data.number)  # type: ignore[attr-defined]
             pr_data.changed_files = get_pr_changed_files(
                 full_pr, github_connection, owner, repo_name
             )

pyproject.toml

@@ -4,9 +4,7 @@ version = "1.0.0"
 description = "Detect pull request conflicts and notify"
 requires-python = ">=3.11"
 dependencies = [
-    "cryptography==49.0.0",
-    "github3-py==4.0.1",
-    "pyjwt==2.13.0",
+    "PyGithub>=2.6.0",
     "python-dotenv==1.2.2",
     "requests==2.34.2",
 ]