github: add GitHubAPI, GitHubAPIClient classes (bug 1989637)

src/lando/main/models/repo.py

@@ -275,6 +275,11 @@ def _github_repo_url(self) -> str | None:
         if self.is_github:
             return self.url.removesuffix(".git")
 
+    @property
+    def _github_repo_org(self) -> str | None:
+        if self.is_github:
+            return self._github_repo_url.split("/")[-2]
+
     @property
     def git_repo_name(self) -> str:
         """Provide the bare name of the Git repo."""

src/lando/main/scm/git.py

@@ -1,4 +1,3 @@
-import asyncio
 import io
 import logging
 import os
@@ -11,8 +10,6 @@
 from pathlib import Path
 from typing import Any
 
-from django.conf import settings
-from simple_github import AppAuth, AppInstallationAuth
 from typing_extensions import override
 
 from lando.main.scm.commit import CommitData
@@ -112,6 +109,8 @@ def push(
         tags: list[str] | None = None,
     ):
         """Push local code to the remote repository."""
+        from lando.utils.github import GitHubAPI
+
         push_command = ["push"]
 
         if force_push:
@@ -134,7 +133,8 @@ def push(
                 repo = match["repo"]
                 repo_name = repo.removesuffix(".git")
 
-                token = self._get_github_token(owner, repo_name)
+                token = GitHubAPI._get_token(owner, repo_name)
+
                 if token:
                     push_path = f"https://git:{token}@github.com/{owner}/{repo}"
 
@@ -152,32 +152,6 @@ def push(
 
         self._git_run(*push_command, cwd=self.path)
 
-    @staticmethod
-    def _get_github_token(repo_owner: str, repo_name: str) -> str | None:
-        """Obtain a fresh GitHub token to push to the specified repo.
-
-        This relies on GITHUB_APP_ID and GITHUB_APP_PRIVKEY to be set in the
-        settings. Returns None if those are missing.
-
-        The app with ID GITHUB_APP_ID needs to be enabled for the target repo.
-
-        """
-        app_id = settings.GITHUB_APP_ID
-        private_key = settings.GITHUB_APP_PRIVKEY
-
-        if not app_id or not private_key:
-            logger.warning(
-                f"Missing GITHUB_APP_ID or GITHUB_APP_PRIVKEY to authenticate against GitHub repo {repo_owner}/{repo_name}",
-            )
-            return None
-
-        app_auth = AppAuth(
-            app_id,
-            private_key,
-        )
-        session = AppInstallationAuth(app_auth, repo_owner, repositories=[repo_name])
-        return asyncio.run(session.get_token())
-
     def last_commit_for_path(self, path: str) -> str:
         """Find last commit to touch a path."""
         command = ["log", "--max-count=1", "--format=%H", "--", path]

src/lando/main/tests/test_git.py

@@ -7,6 +7,7 @@
 from collections.abc import Callable
 from pathlib import Path
 from textwrap import dedent
+from unittest import mock
 from unittest.mock import MagicMock
 
 import pytest
@@ -668,17 +669,27 @@ def test_GitSCM_push(
     )
 
 
-def test_GitSCM_push_get_github_token(git_repo: Path):
+@pytest.fixture
+def mock_github_api_get_token(monkeypatch: pytest.MonkeyPatch):
+    mock_get_token = MagicMock()
+    mock_get_token.side_effect = ["ghs_yolo"]
+
+    monkeypatch.setattr("lando.utils.github.GitHubAPI._get_token", mock_get_token)
+
+    return mock_get_token
+
+
+def test_GitSCM_push_get_github_token(
+    git_repo: Path, mock_github_api_get_token: mock.Mock
+):
     scm = GitSCM(str(git_repo))
     scm._git_run = MagicMock()
-    scm._get_github_token = MagicMock()
-    scm._get_github_token.side_effect = ["ghs_yolo"]
 
     scm.push("https://github.com/some/repo")
 
     assert scm._git_run.call_count == 1, "_git_run wasn't called when pushing"
     assert (
-        scm._get_github_token.call_count == 1
+        mock_github_api_get_token.call_count == 1
     ), "_get_github_token wasn't called when pushing to a github-like URL"
     assert (
         "git:ghs_yolo@github.com" in scm._git_run.call_args[0][1]

src/lando/utils/github.py

@@ -0,0 +1,91 @@
+import asyncio
+import logging
+
+import requests
+from django.conf import settings
+from simple_github import AppAuth, AppInstallationAuth
+
+from lando.main.models.repo import Repo
+
+logger = logging.getLogger(__name__)
+
+
+class GitHubAPI:
+    """A simple wrapper that authenticates with and communicates with the GitHub API."""
+
+    GITHUB_BASE_URL = "https://api.github.com"
+
+    def __init__(self, repo: Repo):
+        repo_owner = repo._github_repo_org
+        repo_name = repo.git_repo_name
+
+        self.session = requests.Session()
+        self.session.headers.update(
+            {
+                "Authorization": f"Bearer {self._get_token(repo_owner, repo_name)}",
+                "User-Agent": settings.HTTP_USER_AGENT,
+                "Accept": "application/vnd.github+json",
+                "X-GitHub-Api-Version": "2022-11-28",
+            }
+        )
+
+    @staticmethod
+    def _get_token(repo_owner: str, repo_name: str) -> str | None:
+        """Obtain a fresh GitHub token to push to the specified repo.
+
+        This relies on GITHUB_APP_ID and GITHUB_APP_PRIVKEY to be set in the
+        environment. Returns None if those are missing.
+
+        The app with ID GITHUB_APP_ID needs to be enabled for the target repo.
+
+        """
+        app_id = settings.GITHUB_APP_ID
+        private_key = settings.GITHUB_APP_PRIVKEY
+
+        if not app_id or not private_key:
+            logger.warning(
+                f"Missing GITHUB_APP_ID or GITHUB_APP_PRIVKEY to authenticate against GitHub repo {repo_owner}/{repo_name}",
+            )
+            return None
+
+        app_auth = AppAuth(
+            app_id,
+            private_key,
+        )
+        session = AppInstallationAuth(app_auth, repo_owner, repositories=[repo_name])
+        return asyncio.run(session.get_token())
+
+    def get(self, path: str, *args, **kwargs) -> dict:
+        """Send a GET request to the GitHub API with given args and kwargs."""
+        url = f"{self.GITHUB_BASE_URL}/{path}"
+        return self.session.get(url, *args, **kwargs)
+
+    def post(self, path: str, *args, **kwargs) -> dict:
+        """Send a POST request to the GitHub API with given args and kwargs."""
+        url = f"self.GITHUB_BASE_URL/{path}"
+        return self.session.post(url, *args, **kwargs)
+
+
+class GitHubAPIClient:
+    """A convenience client that provides various methods to interact with the GitHub API."""
+
+    client = None
+
+    def __init__(self, repo: Repo):
+        self.client = GitHubAPI(repo)
+        self.repo = repo
+        self.repo_base_url = (
+            f"repos/{self.repo._github_repo_org}/{self.repo.git_repo_name}"
+        )
+
+    def _get(self, path: str, *args, **kwargs) -> dict:
+        result = self.client.get(path, *args, **kwargs)
+        return result.json()
+
+    def list_pull_requests(self) -> list:
+        """List all pull requests in the repo."""
+        return self._get(f"{self.repo_base_url}/pulls")
+
+    def get_pull_request(self, pull_number: int) -> dict:
+        """Get a specific pull request from the repo."""
+        return self._get(f"{self.repo_base_url}/pulls/{pull_number}")