ref(scm): search endpoint abstraction (#76627)

src/sentry/integrations/bitbucket/search.py

@@ -1,73 +1,66 @@
 import logging
+from typing import TypeVar
 
-from rest_framework.request import Request
 from rest_framework.response import Response
 
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import control_silo_endpoint
-from sentry.integrations.api.bases.integration import IntegrationEndpoint
 from sentry.integrations.bitbucket.integration import BitbucketIntegration
 from sentry.integrations.models.integration import Integration
+from sentry.integrations.source_code_management.issues import SourceCodeIssueIntegration
+from sentry.integrations.source_code_management.search import SourceCodeSearchEndpoint
 from sentry.shared_integrations.exceptions import ApiError
 
 logger = logging.getLogger("sentry.integrations.bitbucket")
 
+T = TypeVar("T", bound=SourceCodeIssueIntegration)
+
 
 @control_silo_endpoint
-class BitbucketSearchEndpoint(IntegrationEndpoint):
+class BitbucketSearchEndpoint(SourceCodeSearchEndpoint):
     owner = ApiOwner.INTEGRATIONS
     publish_status = {
         "GET": ApiPublishStatus.PRIVATE,
     }
 
-    def get(self, request: Request, organization, integration_id, **kwds) -> Response:
-        try:
-            integration = Integration.objects.get(
-                organizationintegration__organization_id=organization.id,
-                id=integration_id,
-                provider="bitbucket",
-            )
-        except Integration.DoesNotExist:
-            return Response(status=404)
-
-        field = request.GET.get("field")
-        query = request.GET.get("query")
-        if field is None:
-            return Response({"detail": "field is a required parameter"}, status=400)
-        if not query:
-            return Response({"detail": "query is a required parameter"}, status=400)
+    @property
+    def repository_field(self):
+        return "repo"
 
-        installation = integration.get_installation(organization_id=organization.id)
-        assert isinstance(installation, BitbucketIntegration), installation
+    @property
+    def integration_provider(self):
+        return "bitbucket"
 
-        if field == "externalIssue":
-            repo = request.GET.get("repo")
-            if not repo:
-                return Response({"detail": "repo is a required parameter"}, status=400)
+    @property
+    def installation_class(self):
+        return BitbucketIntegration
 
-            full_query = f'title~"{query}"'
-            try:
-                resp = installation.search_issues(query=full_query, repo=repo)
-            except ApiError as e:
-                if "no issue tracker" in str(e):
-                    logger.info(
-                        "bitbucket.issue-search-no-issue-tracker",
-                        extra={"installation_id": installation.model.id, "repo": repo},
-                    )
-                    return Response(
-                        {"detail": "Bitbucket Repository has no issue tracker."}, status=400
-                    )
-                raise
-            return Response(
-                [
-                    {"label": "#{} {}".format(i["id"], i["title"]), "value": i["id"]}
-                    for i in resp.get("values", [])
-                ]
-            )
+    def handle_search_issues(self, installation: T, query: str, repo: str) -> Response:
+        full_query = f'title~"{query}"'
+        try:
+            response = installation.search_issues(query=full_query, repo=repo)
+        except ApiError as e:
+            if "no issue tracker" in str(e):
+                logger.info(
+                    "bitbucket.issue-search-no-issue-tracker",
+                    extra={"installation_id": installation.model.id, "repo": repo},
+                )
+                return Response(
+                    {"detail": "Bitbucket Repository has no issue tracker."}, status=400
+                )
+            raise
 
-        if field == "repo":
-            result = installation.get_repositories(query)
-            return Response([{"label": i["name"], "value": i["name"]} for i in result])
+        assert isinstance(response, dict)
+        return Response(
+            [
+                {"label": "#{} {}".format(i["id"], i["title"]), "value": i["id"]}
+                for i in response.get("values", [])
+            ]
+        )
 
-        return Response(status=400)
+    def handle_search_repositories(
+        self, integration: Integration, installation: T, query: str
+    ) -> Response:
+        result = installation.get_repositories(query)
+        return Response([{"label": i["name"], "value": i["name"]} for i in result])

src/sentry/integrations/github/search.py

@@ -1,84 +1,71 @@
-from typing import Any
+from typing import TypeVar
 
-from rest_framework.request import Request
 from rest_framework.response import Response
 
-from sentry.api.api_owners import ApiOwner
-from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import control_silo_endpoint
-from sentry.integrations.api.bases.integration import IntegrationEndpoint
 from sentry.integrations.github.integration import GitHubIntegration, build_repository_query
 from sentry.integrations.github_enterprise.integration import GitHubEnterpriseIntegration
 from sentry.integrations.models.integration import Integration
-from sentry.organizations.services.organization import RpcOrganization
+from sentry.integrations.source_code_management.issues import SourceCodeIssueIntegration
+from sentry.integrations.source_code_management.search import SourceCodeSearchEndpoint
 from sentry.shared_integrations.exceptions import ApiError
 
+T = TypeVar("T", bound=SourceCodeIssueIntegration)
+
 
 @control_silo_endpoint
-class GithubSharedSearchEndpoint(IntegrationEndpoint):
-    owner = ApiOwner.ECOSYSTEM
-    publish_status = {
-        "GET": ApiPublishStatus.PRIVATE,
-    }
+class GithubSharedSearchEndpoint(SourceCodeSearchEndpoint):
     """NOTE: This endpoint is a shared search endpoint for Github and Github Enterprise integrations."""
 
-    def get(
-        self, request: Request, organization: RpcOrganization, integration_id: int, **kwds: Any
-    ) -> Response:
-        try:
-            integration = Integration.objects.get(
-                organizationintegration__organization_id=organization.id,
-                id=integration_id,
-            )
-        except Integration.DoesNotExist:
-            return Response(status=404)
+    @property
+    def repository_field(self):
+        return "repo"
+
+    @property
+    def integration_provider(self):
+        return None
 
-        field = request.GET.get("field")
-        query = request.GET.get("query")
-        if field is None:
-            return Response({"detail": "field is a required parameter"}, status=400)
-        if not query:
-            return Response({"detail": "query is a required parameter"}, status=400)
+    @property
+    def installation_class(self):
+        return (GitHubIntegration, GitHubEnterpriseIntegration)
 
-        installation = integration.get_installation(organization.id)
-        assert isinstance(
-            installation, (GitHubIntegration, GitHubEnterpriseIntegration)
-        ), installation
-        if field == "externalIssue":
-            repo = request.GET.get("repo")
-            if repo is None:
-                return Response({"detail": "repo is a required parameter"}, status=400)
+    def handle_search_issues(self, installation: T, query: str, repo: str) -> Response:
+        assert isinstance(installation, self.installation_class)
 
-            try:
-                response = installation.search_issues(query=f"repo:{repo} {query}")
-            except ApiError as err:
-                if err.code == 403:
-                    return Response({"detail": "Rate limit exceeded"}, status=429)
-                raise
-            return Response(
-                [
-                    {"label": "#{} {}".format(i["number"], i["title"]), "value": i["number"]}
-                    for i in response.get("items", [])
-                ]
-            )
+        try:
+            response = installation.search_issues(query=f"repo:{repo} {query}")
+        except ApiError as err:
+            if err.code == 403:
+                return Response({"detail": "Rate limit exceeded"}, status=429)
+            raise
+
+        assert isinstance(response, dict)
+        return Response(
+            [
+                {"label": "#{} {}".format(i["number"], i["title"]), "value": i["number"]}
+                for i in response.get("items", [])
+            ]
+        )
 
-        if field == "repo":
-            full_query = build_repository_query(integration.metadata, integration.name, query)
-            try:
-                response = installation.get_client().search_repositories(full_query)
-            except ApiError as err:
-                if err.code == 403:
-                    return Response({"detail": "Rate limit exceeded"}, status=429)
-                if err.code == 422:
-                    return Response(
-                        {
-                            "detail": "Repositories could not be searched because they do not exist, or you do not have access to them."
-                        },
-                        status=404,
-                    )
-                raise
-            return Response(
-                [{"label": i["name"], "value": i["full_name"]} for i in response.get("items", [])]
-            )
+    def handle_search_repositories(
+        self, integration: Integration, installation: T, query: str
+    ) -> Response:
+        assert isinstance(installation, self.installation_class)
 
-        return Response(status=400)
+        full_query = build_repository_query(integration.metadata, integration.name, query)
+        try:
+            response = installation.get_client().search_repositories(full_query)
+        except ApiError as err:
+            if err.code == 403:
+                return Response({"detail": "Rate limit exceeded"}, status=429)
+            if err.code == 422:
+                return Response(
+                    {
+                        "detail": "Repositories could not be searched because they do not exist, or you do not have access to them."
+                    },
+                    status=404,
+                )
+            raise
+        return Response(
+            [{"label": i["name"], "value": i["full_name"]} for i in response.get("items", [])]
+        )

src/sentry/integrations/gitlab/search.py

@@ -1,82 +1,68 @@
-from typing import Any
+from typing import TypeVar
 
-from rest_framework.request import Request
 from rest_framework.response import Response
 
-from sentry.api.api_owners import ApiOwner
-from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import control_silo_endpoint
-from sentry.integrations.api.bases.integration import IntegrationEndpoint
 from sentry.integrations.gitlab.integration import GitlabIntegration
 from sentry.integrations.models.integration import Integration
-from sentry.organizations.services.organization import RpcOrganization
+from sentry.integrations.source_code_management.issues import SourceCodeIssueIntegration
+from sentry.integrations.source_code_management.search import SourceCodeSearchEndpoint
 from sentry.shared_integrations.exceptions import ApiError
 
+T = TypeVar("T", bound=SourceCodeIssueIntegration)
 
-@control_silo_endpoint
-class GitlabIssueSearchEndpoint(IntegrationEndpoint):
-    owner = ApiOwner.INTEGRATIONS
-    publish_status = {
-        "GET": ApiPublishStatus.PRIVATE,
-    }
 
-    def get(
-        self, request: Request, organization: RpcOrganization, integration_id: int, **kwds: Any
-    ) -> Response:
-        try:
-            integration = Integration.objects.get(
-                organizationintegration__organization_id=organization.id,
-                id=integration_id,
-                provider="gitlab",
-            )
-        except Integration.DoesNotExist:
-            return Response(status=404)
+@control_silo_endpoint
+class GitlabIssueSearchEndpoint(SourceCodeSearchEndpoint):
+    @property
+    def repository_field(self):
+        return "project"
 
-        field = request.GET.get("field")
-        query = request.GET.get("query")
-        if field is None:
-            return Response({"detail": "field is a required parameter"}, status=400)
-        if query is None:
-            return Response({"detail": "query is a required parameter"}, status=400)
+    @property
+    def integration_provider(self):
+        return "gitlab"
 
-        installation = integration.get_installation(organization.id)
-        assert isinstance(installation, GitlabIntegration), installation
+    @property
+    def installation_class(self):
+        return GitlabIntegration
 
-        if field == "externalIssue":
-            project = request.GET.get("project")
-            if project is None:
-                return Response({"detail": "project is a required parameter"}, status=400)
-            try:
-                iids = [int(query)]
-                query = None
-            except ValueError:
-                iids = None
+    def handle_search_issues(self, installation: T, query: str, repo: str) -> Response:
+        assert isinstance(installation, self.installation_class)
+        full_query: str | None = query
 
-            try:
-                response = installation.search_issues(query=query, project_id=project, iids=iids)
-            except ApiError as e:
-                return Response({"detail": str(e)}, status=400)
+        try:
+            iids = [int(query)]
+            full_query = None
+        except ValueError:
+            iids = None
 
-            return Response(
-                [
-                    {
-                        "label": "(#{}) {}".format(i["iid"], i["title"]),
-                        "value": "{}#{}".format(i["project_id"], i["iid"]),
-                    }
-                    for i in response
-                ]
-            )
+        try:
+            response = installation.search_issues(query=full_query, project_id=repo, iids=iids)
+        except ApiError as e:
+            return Response({"detail": str(e)}, status=400)
 
-        elif field == "project":
-            try:
-                response = installation.search_projects(query)
-            except ApiError as e:
-                return Response({"detail": str(e)}, status=400)
-            return Response(
-                [
-                    {"label": project["name_with_namespace"], "value": project["id"]}
-                    for project in response
-                ]
-            )
+        assert isinstance(response, list)
+        return Response(
+            [
+                {
+                    "label": "(#{}) {}".format(i["iid"], i["title"]),
+                    "value": "{}#{}".format(i["project_id"], i["iid"]),
+                }
+                for i in response
+            ]
+        )
 
-        return Response({"detail": "invalid field value"}, status=400)
+    def handle_search_repositories(
+        self, integration: Integration, installation: T, query: str
+    ) -> Response:
+        assert isinstance(installation, self.installation_class)
+        try:
+            response = installation.search_projects(query)
+        except ApiError as e:
+            return Response({"detail": str(e)}, status=400)
+        return Response(
+            [
+                {"label": project["name_with_namespace"], "value": project["id"]}
+                for project in response
+            ]
+        )