feat(integrations): Add paginated repo fetching for GitHub integration

src/sentry/integrations/api/endpoints/organization_integration_repos.py

@@ -15,6 +15,7 @@
 from sentry.models.organization import Organization
 from sentry.models.repository import Repository
 from sentry.shared_integrations.exceptions import IntegrationError
+from sentry.utils.cursors import Cursor, CursorResult
 
 
 class IntegrationRepository(TypedDict):
@@ -69,28 +70,81 @@ def get(
         if isinstance(install, RepositoryIntegration):
             search = request.GET.get("search")
             accessible_only = request.GET.get("accessibleOnly", "false").lower() == "true"
+            paginate = request.GET.get("paginate", "false").lower() == "true"
+
+            # Paginated path: opt-in via paginate=true, only when not
+            # searching, and only for integrations that support it.
+            if paginate and not search:
+                result = self._get_paginated_repos(request, install, installed_repo_names)
+                if result is not None:
+                    return result
 
             try:
                 repositories = install.get_repositories(search, accessible_only=accessible_only)
             except (IntegrationError, IdentityNotValid) as e:
                 return self.respond({"detail": str(e)}, status=400)
 
-            installable_only = request.GET.get("installableOnly", "false").lower() == "true"
-
-            # Include a repository if the request is for all repositories, or if we want
-            # installable-only repositories and the repository isn't already installed
-            serialized_repositories = [
-                IntegrationRepository(
-                    name=repo["name"],
-                    identifier=repo["identifier"],
-                    defaultBranch=repo.get("default_branch"),
-                    isInstalled=repo["identifier"] in installed_repo_names,
-                )
-                for repo in repositories
-                if not installable_only or repo["identifier"] not in installed_repo_names
-            ]
             return self.respond(
-                {"repos": serialized_repositories, "searchable": install.repo_search}
+                {
+                    "repos": self._serialize_repos(repositories, installed_repo_names, request),
+                    "searchable": install.repo_search,
+                }
             )
 
         return self.respond({"detail": "Repositories not supported"}, status=400)
+
+    def _get_paginated_repos(
+        self,
+        request: Request,
+        install: RepositoryIntegration,
+        installed_repo_names: set[str],
+    ) -> Response | None:
+        cursor_param = request.GET.get("cursor")
+        try:
+            cursor = Cursor.from_string(cursor_param) if cursor_param else Cursor(0, 0, False)
+            per_page = max(1, min(int(request.GET.get("per_page", 100)), 100))
+        except (ValueError, TypeError):
+            return self.respond({"detail": "Invalid cursor or per_page parameter."}, status=400)
+        page_number = (cursor.offset // per_page) + 1
+
+        try:
+            result = install.get_repositories_page(page=page_number, per_page=per_page)
+        except (IntegrationError, IdentityNotValid) as e:
+            return self.respond({"detail": str(e)}, status=400)
+
+        if result is None:
+            return None
+
+        repositories, has_next = result
+
+        response = self.respond(
+            {
+                "repos": self._serialize_repos(repositories, installed_repo_names, request),
+                "searchable": install.repo_search,
+            }
+        )
+        cursor_result: CursorResult = CursorResult(
+            results=[],
+            prev=Cursor(0, max(0, cursor.offset - per_page), True, cursor.offset > 0),
+            next=Cursor(0, cursor.offset + per_page, False, has_next),
+        )
+        self.add_cursor_headers(request, response, cursor_result)
+        return response
+
+    @staticmethod
+    def _serialize_repos(
+        repositories: list[dict[str, Any]],
+        installed_repo_names: set[str],
+        request: Request,
+    ) -> list[IntegrationRepository]:
+        installable_only = request.GET.get("installableOnly", "false").lower() == "true"
+        return [
+            IntegrationRepository(
+                name=repo["name"],
+                identifier=repo["identifier"],
+                defaultBranch=repo.get("default_branch"),
+                isInstalled=repo["identifier"] in installed_repo_names,
+            )
+            for repo in repositories
+            if not installable_only or repo["identifier"] not in installed_repo_names
+        ]

src/sentry/integrations/github/client.py

@@ -549,6 +549,29 @@ def get_repos(self, page_number_limit: int | None = None) -> list[dict[str, Any]
                 page_number_limit=page_number_limit,
             )
 
+    def get_repos_page(
+        self, page: int = 1, per_page: int | None = None
+    ) -> tuple[list[dict[str, Any]], bool]:
+        """
+        Fetch a single page of repositories accessible to the GitHub App installation.
+        Returns (repositories, has_next_page).
+
+        Unlike get_repos() which aggregates all pages, this returns one page at a time
+        for cursor-based pagination support.
+        """
+        if per_page is None:
+            per_page = self.page_size
+        with SCMIntegrationInteractionEvent(
+            interaction_type=SCMIntegrationInteractionType.GET_REPOSITORIES,
+            provider_key=self.integration_name,
+            integration_id=self.integration.id,
+        ).capture():
+            resp = self.get(
+                "/installation/repositories",
+                params={"per_page": per_page, "page": page},
+            )
+            return resp["repositories"], get_next_link(resp) is not None
+
     def search_repositories(self, query: bytes) -> Mapping[str, Sequence[Any]]:
         """
         Find repositories matching a query.

src/sentry/integrations/github/integration.py

@@ -335,15 +335,7 @@ def get_repositories(
         """
         if not query or accessible_only:
             all_repos = self.get_client().get_repos(page_number_limit=page_number_limit)
-            repos = [
-                {
-                    "name": i["name"],
-                    "identifier": i["full_name"],
-                    "default_branch": i.get("default_branch"),
-                }
-                for i in all_repos
-                if not i.get("archived")
-            ]
+            repos = self._format_repos(all_repos)
             if query:
                 query_lower = query.lower()
                 repos = [r for r in repos if query_lower in r["identifier"].lower()]
@@ -360,6 +352,28 @@ def get_repositories(
             for i in response.get("items", [])
         ]
 
+    def get_repositories_page(
+        self, page: int = 1, per_page: int = 100
+    ) -> tuple[list[dict[str, Any]], bool]:
+        """
+        Fetch a single page of non-archived repositories.
+        Returns (formatted_repos, has_next_page).
+        """
+        raw_repos, has_next = self.get_client().get_repos_page(page=page, per_page=per_page)
+        return self._format_repos(raw_repos), has_next
+
+    @staticmethod
+    def _format_repos(raw_repos: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        return [
+            {
+                "name": i["name"],
+                "identifier": i["full_name"],
+                "default_branch": i.get("default_branch"),
+            }
+            for i in raw_repos
+            if not i.get("archived")
+        ]
+
     def get_unmigratable_repositories(self) -> list[RpcRepository]:
         accessible_repos = self.get_repositories()
         accessible_repo_names = [r["identifier"] for r in accessible_repos]

src/sentry/integrations/source_code_management/repository.py

@@ -60,6 +60,19 @@ def get_repositories(
         """
         raise NotImplementedError
 
+    def get_repositories_page(
+        self,
+        page: int = 1,
+        per_page: int = 100,
+    ) -> tuple[list[dict[str, Any]], bool] | None:
+        """
+        Fetch a single page of repositories. Override to support
+        page-by-page fetching for cursor-based pagination.
+
+        Returns (repos, has_next_page) or None if not supported.
+        """
+        return None
+
 
 class RepositoryIntegration(IntegrationInstallation, BaseRepositoryIntegration, ABC):
     @property

tests/sentry/integrations/api/endpoints/test_organization_integration_repos.py

@@ -262,6 +262,123 @@ def test_accessible_only_with_installable_only(self, get_repositories: MagicMock
             "searchable": True,
         }
 
+    @patch(
+        "sentry.integrations.github.integration.GitHubIntegration.get_repositories_page",
+        return_value=([], False),
+    )
+    def test_paginate_first_page(self, get_repositories_page: MagicMock) -> None:
+        """paginate=true uses get_repositories_page for single-page fetching."""
+        get_repositories_page.return_value = (
+            [
+                {"name": "repo-a", "identifier": "Example/repo-a", "default_branch": "main"},
+                {"name": "repo-b", "identifier": "Example/repo-b", "default_branch": "main"},
+            ],
+            True,
+        )
+        response = self.client.get(self.path, format="json", data={"paginate": "true"})
+
+        assert response.status_code == 200, response.content
+        get_repositories_page.assert_called_once_with(page=1, per_page=100)
+        assert response.data == {
+            "repos": [
+                {
+                    "name": "repo-a",
+                    "identifier": "Example/repo-a",
+                    "defaultBranch": "main",
+                    "isInstalled": False,
+                },
+                {
+                    "name": "repo-b",
+                    "identifier": "Example/repo-b",
+                    "defaultBranch": "main",
+                    "isInstalled": False,
+                },
+            ],
+            "searchable": True,
+        }
+        assert 'results="true"' in response["Link"]
+
+    @patch(
+        "sentry.integrations.github.integration.GitHubIntegration.get_repositories_page",
+        return_value=([], False),
+    )
+    def test_paginate_second_page(self, get_repositories_page: MagicMock) -> None:
+        """Passing a cursor fetches the corresponding page."""
+        get_repositories_page.return_value = (
+            [{"name": "repo-c", "identifier": "Example/repo-c", "default_branch": "main"}],
+            False,
+        )
+        response = self.client.get(
+            self.path, format="json", data={"paginate": "true", "cursor": "0:100:0"}
+        )
+
+        assert response.status_code == 200, response.content
+        get_repositories_page.assert_called_once_with(page=2, per_page=100)
+        # next cursor should indicate no more results
+        assert 'rel="next"; results="false"' in response["Link"]
+        # prev cursor should indicate results exist
+        assert 'rel="previous"; results="true"' in response["Link"]
+
+    @patch(
+        "sentry.integrations.github.integration.GitHubIntegration.get_repositories",
+        return_value=[],
+    )
+    def test_paginate_with_search_falls_through(self, get_repositories: MagicMock) -> None:
+        """paginate=true with search uses the non-paginated path."""
+        get_repositories.return_value = [
+            {"name": "rad-repo", "identifier": "Example/rad-repo", "default_branch": "main"},
+        ]
+        response = self.client.get(
+            self.path, format="json", data={"paginate": "true", "search": "rad"}
+        )
+
+        assert response.status_code == 200, response.content
+        get_repositories.assert_called_once()
+        assert "Link" not in response
+
+    @patch(
+        "sentry.integrations.github.integration.GitHubIntegration.get_repositories",
+        return_value=[],
+    )
+    def test_no_paginate_param_uses_existing_path(self, get_repositories: MagicMock) -> None:
+        """Without paginate=true, get_repositories is called (not get_repositories_page)."""
+        get_repositories.return_value = [
+            {"name": "rad-repo", "identifier": "Example/rad-repo", "default_branch": "main"},
+        ]
+        response = self.client.get(self.path, format="json")
+
+        assert response.status_code == 200, response.content
+        get_repositories.assert_called_once()
+        assert "Link" not in response
+
+    @patch(
+        "sentry.integrations.github.integration.GitHubIntegration.get_repositories_page",
+        return_value=([], False),
+    )
+    def test_paginate_installable_only(self, get_repositories_page: MagicMock) -> None:
+        """installableOnly filter works with the paginated path."""
+        get_repositories_page.return_value = (
+            [
+                {"name": "installed", "identifier": "Example/installed", "default_branch": "main"},
+                {"name": "new-repo", "identifier": "Example/new-repo", "default_branch": "main"},
+            ],
+            False,
+        )
+        self.create_repo(
+            project=self.project,
+            integration_id=self.integration.id,
+            name="Example/installed",
+        )
+        response = self.client.get(
+            self.path,
+            format="json",
+            data={"paginate": "true", "installableOnly": "true"},
+        )
+
+        assert response.status_code == 200, response.content
+        assert len(response.data["repos"]) == 1
+        assert response.data["repos"][0]["identifier"] == "Example/new-repo"
+
     def test_no_repository_method(self) -> None:
         integration = self.create_integration(
             organization=self.org, provider="jira", name="Example", external_id="example:1"