types(py): Github (#30006)

mypy.ini

@@ -34,6 +34,7 @@ files = src/sentry/api/bases/external_actor.py,
         src/sentry/grouping/strategies/template.py,
         src/sentry/grouping/strategies/utils.py,
         src/sentry/integrations/base.py,
+        src/sentry/integrations/github/,
         src/sentry/integrations/slack/,
         src/sentry/integrations/vsts/,
         src/sentry/killswitches.py,

src/sentry/integrations/bitbucket/issues.py

@@ -1,3 +1,5 @@
+from typing import Sequence
+
 from django.urls import reverse
 
 from sentry.integrations.issues import IssueBasicMixin
@@ -24,7 +26,7 @@ class BitbucketIssueBasicMixin(IssueBasicMixin):
         repo, issue_id = key.split("#")
         return f"https://bitbucket.org/{repo}/issues/{issue_id}"
 
-    def get_persisted_default_config_fields(self):
+    def get_persisted_default_config_fields(self) -> Sequence[str]:
         return ["repo"]
 
     def get_create_issue_config(self, group, user, **kwargs):
@@ -37,6 +39,7 @@ class BitbucketIssueBasicMixin(IssueBasicMixin):
             "sentry-extensions-bitbucket-search", args=[org.slug, self.model.id]
         )
 
+        # TODO(mgaeta): inline these lists.
         return (
             [
                 {
@@ -119,7 +122,7 @@ class BitbucketIssueBasicMixin(IssueBasicMixin):
         try:
             issue = client.create_issue(data.get("repo"), data)
         except ApiError as e:
-            self.raise_error(e)
+            raise self.raise_error(e)
 
         return {
             "key": issue["id"],

src/sentry/integrations/example/integration.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Mapping
+from typing import Any, Mapping, Sequence
 
 from django.http import HttpResponse
 
@@ -76,7 +76,7 @@ class ExampleIntegration(IntegrationInstallation, IssueSyncMixin):
         }
         return comment
 
-    def get_persisted_default_config_fields(self):
+    def get_persisted_default_config_fields(self) -> Sequence[str]:
         return ["project", "issueType"]
 
     def get_persisted_user_default_config_fields(self):

src/sentry/integrations/github/__init__.py

@@ -1,3 +1,3 @@
 from sentry.utils.imports import import_submodules
 
-import_submodules(globals(), __name__, __path__)
+import_submodules(globals(), __name__, __path__)  # type: ignore

src/sentry/integrations/github/client.py

@@ -1,61 +1,91 @@
 from __future__ import annotations
 
 from datetime import datetime
+from typing import Any, Mapping, Sequence
 
 import sentry_sdk
+from rest_framework.response import Response
 
 from sentry.integrations.client import ApiClient
 from sentry.integrations.github.utils import get_jwt
-from sentry.models import Repository
+from sentry.models import Integration, Repository
 from sentry.utils import jwt
+from sentry.utils.json import JSONData
 
 
-class GitHubClientMixin(ApiClient):
+class GitHubClientMixin(ApiClient):  # type: ignore
     allow_redirects = True
 
     base_url = "https://api.github.com"
     integration_name = "github"
 
-    def get_jwt(self):
+    def get_jwt(self) -> str:
         return get_jwt()
 
-    def get_last_commits(self, repo, end_sha):
-        # return api request that fetches last ~30 commits
-        # see https://developer.github.com/v3/repos/commits/#list-commits-on-a-repository
-        # using end_sha as parameter
-        return self.get_cached(f"/repos/{repo}/commits", params={"sha": end_sha})
-
-    def compare_commits(self, repo, start_sha, end_sha):
-        # see https://developer.github.com/v3/repos/commits/#compare-two-commits
-        # where start sha is oldest and end is most recent
-        return self.get_cached(f"/repos/{repo}/compare/{start_sha}...{end_sha}")
-
-    def repo_hooks(self, repo):
-        return self.get(f"/repos/{repo}/hooks")
-
-    def get_commits(self, repo):
-        return self.get(f"/repos/{repo}/commits")
-
-    def get_commit(self, repo, sha):
-        return self.get_cached(f"/repos/{repo}/commits/{sha}")
-
-    def get_repo(self, repo):
-        return self.get(f"/repos/{repo}")
+    def get_last_commits(self, repo: str, end_sha: str) -> Sequence[JSONData]:
+        """
+        Return API request that fetches last ~30 commits
+        see https://developer.github.com/v3/repos/commits/#list-commits-on-a-repository
+        using end_sha as parameter.
+        """
+        # Explicitly typing to satisfy mypy.
+        commits: Sequence[JSONData] = self.get_cached(
+            f"/repos/{repo}/commits", params={"sha": end_sha}
+        )
+        return commits
 
-    def get_repositories(self):
-        repositories = self.get("/installation/repositories", params={"per_page": 100})
+    def compare_commits(self, repo: str, start_sha: str, end_sha: str) -> JSONData:
+        """
+        See https://developer.github.com/v3/repos/commits/#compare-two-commits
+        where start sha is oldest and end is most recent.
+        """
+        # Explicitly typing to satisfy mypy.
+        diff: JSONData = self.get_cached(f"/repos/{repo}/compare/{start_sha}...{end_sha}")
+        return diff
+
+    def repo_hooks(self, repo: str) -> Sequence[JSONData]:
+        # Explicitly typing to satisfy mypy.
+        hooks: Sequence[JSONData] = self.get(f"/repos/{repo}/hooks")
+        return hooks
+
+    def get_commits(self, repo: str) -> Sequence[JSONData]:
+        # Explicitly typing to satisfy mypy.
+        commits: Sequence[JSONData] = self.get(f"/repos/{repo}/commits")
+        return commits
+
+    def get_commit(self, repo: str, sha: str) -> JSONData:
+        # Explicitly typing to satisfy mypy.
+        commit: JSONData = self.get_cached(f"/repos/{repo}/commits/{sha}")
+        return commit
+
+    def get_repo(self, repo: str) -> JSONData:
+        # Explicitly typing to satisfy mypy.
+        repository: JSONData = self.get(f"/repos/{repo}")
+        return repository
+
+    def get_repositories(self) -> Sequence[JSONData]:
+        # Explicitly typing to satisfy mypy.
+        repositories: JSONData = self.get("/installation/repositories", params={"per_page": 100})
         repos = repositories["repositories"]
         return [repo for repo in repos if not repo.get("archived")]
 
-    def search_repositories(self, query):
-        return self.get("/search/repositories", params={"q": query})
+    def search_repositories(self, query: bytes) -> Mapping[str, Sequence[JSONData]]:
+        # Explicitly typing to satisfy mypy.
+        repositories: Mapping[str, Sequence[JSONData]] = self.get(
+            "/search/repositories", params={"q": query}
+        )
+        return repositories
 
-    def get_assignees(self, repo):
-        return self.get_with_pagination(f"/repos/{repo}/assignees")
+    def get_assignees(self, repo: str) -> Sequence[JSONData]:
+        # Explicitly typing to satisfy mypy.
+        assignees: Sequence[JSONData] = self.get_with_pagination(f"/repos/{repo}/assignees")
+        return assignees
 
-    def get_with_pagination(self, path, *args, **kwargs):
+    def get_with_pagination(self, path: str, *args: Any, **kwargs: Any) -> Sequence[JSONData]:
         """
-        Github uses the Link header to provide pagination links. Github recommends using the provided link relations and not constructing our own URL.
+        Github uses the Link header to provide pagination links. Github
+        recommends using the provided link relations and not constructing our
+        own URL.
         https://docs.github.com/en/rest/guides/traversing-with-pagination
         """
         try:
@@ -78,18 +108,21 @@ class GitHubClientMixin(ApiClient):
             output.extend(resp)
             page_number = 1
 
-            def get_next_link(resp):
-                link = resp.headers.get("link")
-                if link is None:
+            # TODO(mgaeta): Move this to utils.
+            def get_next_link(resp: Response) -> str | None:
+                link_option: str | None = resp.headers.get("link")
+                if link_option is None:
                     return None
 
                 # Should be a comma separated string of links
-                links = link.split(",")
+                links = link_option.split(",")
 
                 for link in links:
                     # If there is a 'next' link return the URL between the angle brackets, or None
                     if 'rel="next"' in link:
-                        return link[link.find("<") + 1 : link.find(">")]
+                        start = link.find("<") + 1
+                        end = link.find(">")
+                        return link[start:end]
 
                 return None
 
@@ -99,27 +132,39 @@ class GitHubClientMixin(ApiClient):
                 page_number += 1
             return output
 
-    def get_issues(self, repo):
-        return self.get(f"/repos/{repo}/issues")
+    def get_issues(self, repo: str) -> Sequence[JSONData]:
+        issues: Sequence[JSONData] = self.get(f"/repos/{repo}/issues")
+        return issues
 
-    def search_issues(self, query):
-        return self.get("/search/issues", params={"q": query})
+    def search_issues(self, query: str) -> Mapping[str, Sequence[Mapping[str, Any]]]:
+        # Explicitly typing to satisfy mypy.
+        issues: Mapping[str, Sequence[Mapping[str, Any]]] = self.get(
+            "/search/issues", params={"q": query}
+        )
+        return issues
 
-    def get_issue(self, repo, number):
+    def get_issue(self, repo: str, number: str) -> JSONData:
         return self.get(f"/repos/{repo}/issues/{number}")
 
-    def create_issue(self, repo, data):
+    def create_issue(self, repo: str, data: Mapping[str, Any]) -> JSONData:
         endpoint = f"/repos/{repo}/issues"
         return self.post(endpoint, data=data)
 
-    def create_comment(self, repo, issue_id, data):
+    def create_comment(self, repo: str, issue_id: str, data: Mapping[str, Any]) -> JSONData:
         endpoint = f"/repos/{repo}/issues/{issue_id}/comments"
         return self.post(endpoint, data=data)
 
-    def get_user(self, gh_username):
+    def get_user(self, gh_username: str) -> JSONData:
         return self.get(f"/users/{gh_username}")
 
-    def request(self, method, path, headers=None, data=None, params=None):
+    def request(
+        self,
+        method: str,
+        path: str,
+        headers: Mapping[str, Any] | None = None,
+        data: Mapping[str, Any] | None = None,
+        params: Mapping[str, Any] | None = None,
+    ) -> JSONData:
         if headers is None:
             headers = {
                 "Authorization": f"token {self.get_token()}",
@@ -128,31 +173,31 @@ class GitHubClientMixin(ApiClient):
             }
         return self._request(method, path, headers=headers, data=data, params=params)
 
-    def get_token(self, force_refresh=False):
+    def get_token(self, force_refresh: bool = False) -> str:
         """
         Get token retrieves the active access token from the integration model.
         Should the token have expired, a new token will be generated and
         automatically persisted into the integration.
         """
-        token = self.integration.metadata.get("access_token")
-        expires_at = self.integration.metadata.get("expires_at")
-
-        if expires_at is not None:
-            expires_at = datetime.strptime(expires_at, "%Y-%m-%dT%H:%M:%S")
-
-        if not token or expires_at < datetime.utcnow() or force_refresh:
+        token: str | None = self.integration.metadata.get("access_token")
+        expires_at: str | None = self.integration.metadata.get("expires_at")
+
+        if (
+            not token
+            or not expires_at
+            or (datetime.strptime(expires_at, "%Y-%m-%dT%H:%M:%S") < datetime.utcnow())
+            or force_refresh
+        ):
             res = self.create_token()
             token = res["token"]
-            expires_at = datetime.strptime(res["expires_at"], "%Y-%m-%dT%H:%M:%SZ")
+            expires_at = datetime.strptime(res["expires_at"], "%Y-%m-%dT%H:%M:%SZ").isoformat()
 
-            self.integration.metadata.update(
-                {"access_token": token, "expires_at": expires_at.isoformat()}
-            )
+            self.integration.metadata.update({"access_token": token, "expires_at": expires_at})
             self.integration.save()
 
-        return token
+        return token or ""
 
-    def create_token(self):
+    def create_token(self) -> JSONData:
         headers = {
             # TODO(jess): remove this whenever it's out of preview
             "Accept": "application/vnd.github.machine-man-preview+json",
@@ -164,15 +209,19 @@ class GitHubClientMixin(ApiClient):
         )
 
     def check_file(self, repo: Repository, path: str, version: str) -> str | None:
-        repo_name = repo.name
-        return self.head_cached(path=f"/repos/{repo_name}/contents/{path}", params={"ref": version})
+        file: str = self.head_cached(
+            path=f"/repos/{repo.name}/contents/{path}", params={"ref": version}
+        )
+        return file
 
-    def search_file(self, repo, filename):
+    def search_file(
+        self, repo: Repository, filename: str
+    ) -> Mapping[str, Sequence[Mapping[str, Any]]]:
         query = f"filename:{filename}+repo:{repo}"
-        results = self.get(path="/search/code", params={"q": query})
+        results: Mapping[str, Any] = self.get(path="/search/code", params={"q": query})
         return results
 
-    def get_file(self, repo, path):
+    def get_file(self, repo: str, path: str) -> bytes:
         from base64 import b64decode
 
         # default ref will be the default_branch
@@ -182,6 +231,6 @@ class GitHubClientMixin(ApiClient):
 
 
 class GitHubAppsClient(GitHubClientMixin):
-    def __init__(self, integration):
+    def __init__(self, integration: Integration) -> None:
         self.integration = integration
         super().__init__()

src/sentry/integrations/github/integration.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
 import re
+from typing import Any, Mapping, Sequence
 
 from django.utils.text import slugify
 from django.utils.translation import ugettext_lazy as _
+from rest_framework.request import Request
+from rest_framework.response import Response
 
 from sentry import options
 from sentry.integrations import (
@@ -14,15 +17,15 @@ from sentry.integrations import (
     IntegrationProvider,
 )
 from sentry.integrations.repositories import RepositoryMixin
-from sentry.models import Integration, OrganizationIntegration, Repository
-from sentry.pipeline import PipelineView
+from sentry.models import Integration, Organization, OrganizationIntegration, Repository
+from sentry.pipeline import Pipeline, PipelineView
 from sentry.shared_integrations.constants import ERR_INTERNAL, ERR_UNAUTHORIZED
 from sentry.shared_integrations.exceptions import ApiError
 from sentry.tasks.integrations import migrate_repo
 from sentry.utils import jwt
 from sentry.web.helpers import render_to_response
 
-from .client import GitHubAppsClient
+from .client import GitHubAppsClient, GitHubClientMixin
 from .issues import GitHubIssueBasic
 from .repository import GitHubRepositoryProvider
 from .utils import get_jwt
@@ -60,7 +63,6 @@ FEATURES = [
     ),
 ]
 
-
 metadata = IntegrationMetadata(
     description=DESCRIPTION.strip(),
     features=FEATURES,
@@ -79,20 +81,23 @@ API_ERRORS = {
 }
 
 
-def build_repository_query(metadata, name, query):
+def build_repository_query(metadata: Mapping[str, Any], name: str, query: str) -> bytes:
     account_type = "user" if metadata["account_type"] == "User" else "org"
-    return (f"{account_type}:{name} {query}").encode()
+    return f"{account_type}:{name} {query}".encode()
 
 
-class GitHubIntegration(IntegrationInstallation, GitHubIssueBasic, RepositoryMixin):
+class GitHubIntegration(IntegrationInstallation, GitHubIssueBasic, RepositoryMixin):  # type: ignore
     repo_search = True
 
-    def get_client(self):
+    def get_client(self) -> GitHubClientMixin:
         return GitHubAppsClient(integration=self.model)
 
-    def get_codeowner_file(self, repo, ref=None):
+    def get_codeowner_file(
+        self, repo: Repository, ref: str | None = None
+    ) -> Mapping[str, Any] | None:
         try:
             files = self.get_client().search_file(repo.name, "CODEOWNERS")
+            # TODO(mgaeta): Pull this logic out of the try/catch.
             for f in files["items"]:
                 if f["name"] == "CODEOWNERS":
                     filepath = f["path"]
@@ -101,10 +106,9 @@ class GitHubIntegration(IntegrationInstallation, GitHubIssueBasic, RepositoryMix
                     return {"filepath": filepath, "html_url": html_url, "raw": contents}
         except ApiError:
             return None
-
         return None
 
-    def get_repositories(self, query=None):
+    def get_repositories(self, query: str | None = None) -> Sequence[Mapping[str, Any]]:
         if not query:
             return [
                 {"name": i["name"], "identifier": i["full_name"]}
@@ -117,7 +121,7 @@ class GitHubIntegration(IntegrationInstallation, GitHubIssueBasic, RepositoryMix
             {"name": i["name"], "identifier": i["full_name"]} for i in response.get("items", [])
         ]
 
-    def search_issues(self, query):
+    def search_issues(self, query: str) -> Mapping[str, Sequence[Mapping[str, Any]]]:
         return self.get_client().search_issues(query)
 
     def format_source_url(self, repo: Repository, filepath: str, branch: str) -> str:
@@ -125,7 +129,7 @@ class GitHubIntegration(IntegrationInstallation, GitHubIssueBasic, RepositoryMix
         # "https://github.com/octokit/octokit.rb/blob/master/README.md"
         return f"https://github.com/{repo.name}/blob/{branch}/{filepath}"
 
-    def get_unmigratable_repositories(self):
+    def get_unmigratable_repositories(self) -> Sequence[Repository]:
         accessible_repos = self.get_repositories()
         accessible_repo_names = [r["identifier"] for r in accessible_repos]
 
@@ -135,22 +139,23 @@ class GitHubIntegration(IntegrationInstallation, GitHubIssueBasic, RepositoryMix
 
         return [repo for repo in existing_repos if repo.name not in accessible_repo_names]
 
-    def reinstall(self):
+    def reinstall(self) -> None:
         self.reinstall_repositories()
 
-    def message_from_error(self, exc):
+    def message_from_error(self, exc: Exception) -> str:
+        # TODO(mgaeta): Clean up the conditional flow.
         if isinstance(exc, ApiError):
-            message = API_ERRORS.get(exc.code)
+            message = API_ERRORS.get(exc.code, "")
             if exc.code == 404 and re.search(r"/repos/.*/(compare|commits)", exc.url):
                 message += f" Please also confirm that the commits associated with the following URL have been pushed to GitHub: {exc.url}"
 
-            if message is None:
+            if not message:
                 message = exc.json.get("message", "unknown error") if exc.json else "unknown error"
             return f"Error Communicating with GitHub (HTTP {exc.code}): {message}"
         else:
             return ERR_INTERNAL
 
-    def has_repo_access(self, repo):
+    def has_repo_access(self, repo: Repository) -> bool:
         client = self.get_client()
         try:
             # make sure installation has access to this specific repo
@@ -163,7 +168,7 @@ class GitHubIntegration(IntegrationInstallation, GitHubIssueBasic, RepositoryMix
         return True
 
 
-class GitHubIntegrationProvider(IntegrationProvider):
+class GitHubIntegrationProvider(IntegrationProvider):  # type: ignore
     key = "github"
     name = "GitHub"
     metadata = metadata
@@ -179,10 +184,15 @@ class GitHubIntegrationProvider(IntegrationProvider):
 
     setup_dialog_config = {"width": 1030, "height": 1000}
 
-    def get_client(self):
+    def get_client(self) -> GitHubClientMixin:
         return GitHubAppsClient(integration=self.integration_cls)
 
-    def post_install(self, integration, organization, extra=None):
+    def post_install(
+        self,
+        integration: Integration,
+        organization: Organization,
+        extra: Mapping[str, Any] | None = None,
+    ) -> None:
         repo_ids = Repository.objects.filter(
             organization_id=organization.id,
             provider__in=["github", "integrations:github"],
@@ -198,21 +208,23 @@ class GitHubIntegrationProvider(IntegrationProvider):
                 }
             )
 
-    def get_pipeline_views(self):
+    def get_pipeline_views(self) -> Sequence[PipelineView]:
         return [GitHubInstallationRedirect()]
 
-    def get_installation_info(self, installation_id):
+    def get_installation_info(self, installation_id: str) -> Mapping[str, Any]:
         client = self.get_client()
         headers = {
             # TODO(jess): remove this whenever it's out of preview
             "Accept": "application/vnd.github.machine-man-preview+json",
         }
         headers.update(jwt.authorization_header(get_jwt()))
-        resp = client.get(f"/app/installations/{installation_id}", headers=headers)
 
+        resp: Mapping[str, Any] = client.get(
+            f"/app/installations/{installation_id}", headers=headers
+        )
         return resp
 
-    def build_integration(self, state):
+    def build_integration(self, state: Mapping[str, str]) -> Mapping[str, Any]:
         installation = self.get_installation_info(state["installation_id"])
 
         integration = {
@@ -237,7 +249,7 @@ class GitHubIntegrationProvider(IntegrationProvider):
 
         return integration
 
-    def setup(self):
+    def setup(self) -> None:
         from sentry.plugins.base import bindings
 
         bindings.add(
@@ -245,12 +257,12 @@ class GitHubIntegrationProvider(IntegrationProvider):
         )
 
 
-class GitHubInstallationRedirect(PipelineView):
-    def get_app_url(self):
+class GitHubInstallationRedirect(PipelineView):  # type: ignore
+    def get_app_url(self) -> str:
         name = options.get("github-app.name")
         return f"https://github.com/apps/{slugify(name)}"
 
-    def dispatch(self, request, pipeline):
+    def dispatch(self, request: Request, pipeline: Pipeline) -> Response:
         if "reinstall_id" in request.GET:
             pipeline.bind_state("reinstall_id", request.GET["reinstall_id"])
 

src/sentry/integrations/github/issues.py

@@ -1,20 +1,25 @@
+from __future__ import annotations
+
+from typing import Any, Mapping, Sequence
+
 from django.urls import reverse
 
 from sentry.integrations.issues import IssueBasicMixin
+from sentry.models import ExternalIssue, Group, User
 from sentry.shared_integrations.exceptions import ApiError, IntegrationError
 from sentry.utils.http import absolute_uri
 
 
-class GitHubIssueBasic(IssueBasicMixin):
-    def make_external_key(self, data):
+class GitHubIssueBasic(IssueBasicMixin):  # type: ignore
+    def make_external_key(self, data: Mapping[str, Any]) -> str:
         return "{}#{}".format(data["repo"], data["key"])
 
-    def get_issue_url(self, key):
+    def get_issue_url(self, key: str) -> str:
         domain_name, user = self.model.metadata["domain_name"].split("/")
         repo, issue_id = key.split("#")
         return f"https://{domain_name}/{repo}/issues/{issue_id}"
 
-    def after_link_issue(self, external_issue, **kwargs):
+    def after_link_issue(self, external_issue: ExternalIssue, **kwargs: Any) -> None:
         data = kwargs["data"]
         client = self.get_client()
 
@@ -32,13 +37,15 @@ class GitHubIssueBasic(IssueBasicMixin):
             except ApiError as e:
                 raise IntegrationError(self.message_from_error(e))
 
-    def get_persisted_default_config_fields(self):
+    def get_persisted_default_config_fields(self) -> Sequence[str]:
         return ["repo"]
 
-    def create_default_repo_choice(self, default_repo):
-        return (default_repo, default_repo.split("/")[1])
+    def create_default_repo_choice(self, default_repo: str) -> tuple[str, str]:
+        return default_repo, default_repo.split("/")[1]
 
-    def get_create_issue_config(self, group, user, **kwargs):
+    def get_create_issue_config(
+        self, group: Group, user: User, **kwargs: Any
+    ) -> Sequence[Mapping[str, Any]]:
         kwargs["link_referrer"] = "github_integration"
         fields = super().get_create_issue_config(group, user, **kwargs)
         default_repo, repo_choices = self.get_repository_choices(group, **kwargs)
@@ -50,7 +57,8 @@ class GitHubIssueBasic(IssueBasicMixin):
             "sentry-extensions-github-search", args=[org.slug, self.model.id]
         )
 
-        return (
+        # TODO(mgaeta): inline these lists.
+        out: Sequence[Mapping[str, Any]] = (
             [
                 {
                     "name": "repo",
@@ -75,8 +83,9 @@ class GitHubIssueBasic(IssueBasicMixin):
                 }
             ]
         )
+        return out
 
-    def create_issue(self, data, **kwargs):
+    def create_issue(self, data: Mapping[str, Any], **kwargs: Any) -> Mapping[str, Any]:
         client = self.get_client()
 
         repo = data.get("repo")
@@ -104,7 +113,7 @@ class GitHubIssueBasic(IssueBasicMixin):
             "repo": repo,
         }
 
-    def get_link_issue_config(self, group, **kwargs):
+    def get_link_issue_config(self, group: Group, **kwargs: Any) -> Sequence[Mapping[str, Any]]:
         default_repo, repo_choices = self.get_repository_choices(group, **kwargs)
 
         org = group.organization
@@ -143,11 +152,11 @@ class GitHubIssueBasic(IssueBasicMixin):
                 ),
                 "type": "textarea",
                 "required": False,
-                "help": ("Leave blank if you don't want to " "add a comment to the GitHub issue."),
+                "help": "Leave blank if you don't want to add a comment to the GitHub issue.",
             },
         ]
 
-    def get_issue(self, issue_id, **kwargs):
+    def get_issue(self, issue_id: str, **kwargs: Any) -> Mapping[str, Any]:
         data = kwargs["data"]
         repo = data.get("repo")
         issue_num = data.get("externalIssue")
@@ -172,23 +181,23 @@ class GitHubIssueBasic(IssueBasicMixin):
             "repo": repo,
         }
 
-    def get_allowed_assignees(self, repo):
+    def get_allowed_assignees(self, repo: str) -> Sequence[tuple[str, str]]:
         client = self.get_client()
         try:
             response = client.get_assignees(repo)
         except Exception as e:
-            self.raise_error(e)
+            raise self.raise_error(e)
 
         users = tuple((u["login"], u["login"]) for u in response)
 
         return (("", "Unassigned"),) + users
 
-    def get_repo_issues(self, repo):
+    def get_repo_issues(self, repo: str) -> Sequence[tuple[str, str]]:
         client = self.get_client()
         try:
             response = client.get_issues(repo)
         except Exception as e:
-            self.raise_error(e)
+            raise self.raise_error(e)
 
         issues = tuple((i["number"], "#{} {}".format(i["number"], i["title"])) for i in response)
 

src/sentry/integrations/github/repository.py

@@ -1,19 +1,23 @@
 from __future__ import annotations
 
-from typing import Any, Mapping, MutableMapping
+from typing import Any, Mapping, MutableMapping, Sequence
 
-from sentry.models import Integration, Organization
+from sentry.integrations import IntegrationInstallation
+from sentry.models import Integration, Organization, PullRequest, Repository
 from sentry.plugins.providers import IntegrationRepositoryProvider
 from sentry.shared_integrations.exceptions import ApiError, IntegrationError
+from sentry.utils.json import JSONData
 
 WEBHOOK_EVENTS = ["push", "pull_request"]
 
 
-class GitHubRepositoryProvider(IntegrationRepositoryProvider):
+class GitHubRepositoryProvider(IntegrationRepositoryProvider):  # type: ignore
     name = "GitHub"
     repo_provider = "github"
 
-    def _validate_repo(self, client, installation, repo):
+    def _validate_repo(
+        self, client: Any, installation: IntegrationInstallation, repo: str
+    ) -> JSONData:
         try:
             repo_data = client.get_repo(repo)
         except Exception as e:
@@ -42,7 +46,9 @@ class GitHubRepositoryProvider(IntegrationRepositoryProvider):
 
         return config
 
-    def build_repository_config(self, organization, data):
+    def build_repository_config(
+        self, organization: Organization, data: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
         return {
             "name": data["identifier"],
             "external_id": data["external_id"],
@@ -51,8 +57,10 @@ class GitHubRepositoryProvider(IntegrationRepositoryProvider):
             "integration_id": data["integration_id"],
         }
 
-    def compare_commits(self, repo, start_sha, end_sha):
-        def eval_commits(client):
+    def compare_commits(
+        self, repo: Repository, start_sha: str | None, end_sha: str
+    ) -> Sequence[Mapping[str, Any]]:
+        def eval_commits(client: Any) -> Sequence[Mapping[str, Any]]:
             # use config name because that is kept in sync via webhooks
             name = repo.config["name"]
             if start_sha is None:
@@ -78,11 +86,16 @@ class GitHubRepositoryProvider(IntegrationRepositoryProvider):
                     return eval_commits(client)
                 except Exception as e:
                     raise installation.raise_error(e)
-            installation.raise_error(e)
+            raise installation.raise_error(e)
         except Exception as e:
-            installation.raise_error(e)
+            raise installation.raise_error(e)
 
-    def _format_commits(self, client, repo_name, commit_list):
+    def _format_commits(
+        self,
+        client: Any,
+        repo_name: str,
+        commit_list: Sequence[Mapping[str, Any]],
+    ) -> Sequence[Mapping[str, Any]]:
         """Convert GitHub commits into our internal format
 
         For each commit in the list we have to fetch patch data, as the
@@ -105,12 +118,12 @@ class GitHubRepositoryProvider(IntegrationRepositoryProvider):
             for c in commit_list
         ]
 
-    def _get_patchset(self, client, repo_name, sha):
+    def _get_patchset(self, client: Any, repo_name: str, sha: str) -> Sequence[Mapping[str, Any]]:
         """Get the modified files for a commit"""
         commit = client.get_commit(repo_name, sha)
         return self._transform_patchset(commit["files"])
 
-    def _transform_patchset(self, diff):
+    def _transform_patchset(self, diff: Sequence[Mapping[str, Any]]) -> Sequence[Mapping[str, Any]]:
         """Convert the patch data from GitHub into our internal format
 
         See sentry.models.Release.set_commits
@@ -128,8 +141,10 @@ class GitHubRepositoryProvider(IntegrationRepositoryProvider):
                 changes.append({"path": change["filename"], "type": "A"})
         return changes
 
-    def pull_request_url(self, repo, pull_request):
+    def pull_request_url(self, repo: Repository, pull_request: PullRequest) -> str:
         return f"{repo.url}/pull/{pull_request.key}"
 
-    def repository_external_slug(self, repo):
-        return repo.name
+    def repository_external_slug(self, repo: Repository) -> str:
+        # Explicitly typing to satisfy mypy.
+        slug: str = repo.name
+        return slug

src/sentry/integrations/github/search.py

@@ -1,13 +1,14 @@
+from rest_framework.request import Request
 from rest_framework.response import Response
 
 from sentry.api.bases.integration import IntegrationEndpoint
 from sentry.integrations.github.integration import build_repository_query
-from sentry.models import Integration
+from sentry.models import Integration, Organization
 from sentry.shared_integrations.exceptions import ApiError
 
 
-class GitHubSearchEndpoint(IntegrationEndpoint):
-    def get(self, request, organization, integration_id):
+class GitHubSearchEndpoint(IntegrationEndpoint):  # type: ignore
+    def get(self, request: Request, organization: Organization, integration_id: int) -> Response:
         try:
             integration = Integration.objects.get(organizations=organization, id=integration_id)
         except Integration.DoesNotExist:

src/sentry/integrations/github/utils.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import calendar
 import datetime
 import time
@@ -6,13 +8,13 @@ from sentry import options
 from sentry.utils import jwt
 
 
-def get_jwt(github_id=None, github_private_key=None):
+def get_jwt(github_id: str | None = None, github_private_key: str | None = None) -> str:
     if github_id is None:
         github_id = options.get("github-app.id")
     if github_private_key is None:
         github_private_key = options.get("github-app.private-key")
-    exp = datetime.datetime.utcnow() + datetime.timedelta(minutes=10)
-    exp = calendar.timegm(exp.timetuple())
+    exp_ = datetime.datetime.utcnow() + datetime.timedelta(minutes=10)
+    exp = calendar.timegm(exp_.timetuple())
     # Generate the JWT
     payload = {
         # issued at time