feat(integrations): add option to filter by integration type

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

@@ -21,6 +21,7 @@
     OrganizationIntegrationBaseEndpoint,
 )
 from sentry.integrations.api.serializers.models.integration import OrganizationIntegrationResponse
+from sentry.integrations.base import INTEGRATION_TYPE_TO_PROVIDER
 from sentry.integrations.models.integration import Integration
 from sentry.integrations.models.organization_integration import OrganizationIntegration
 from sentry.organizations.services.organization.model import (
@@ -93,6 +94,7 @@ def get(
         if provider_key is None:
             provider_key = request.GET.get("provider_key", "")
         include_config_raw = request.GET.get("includeConfig")
+        integration_type = request.GET.get("integrationType")
 
         # Include the configurations by default if includeConfig is not present.
         # TODO(mgaeta): HACK. We need a consistent way to get booleans from query parameters.
@@ -109,6 +111,18 @@ def get(
         if provider_key:
             queryset = queryset.filter(integration__provider=provider_key.lower())
 
+        if integration_type:
+            if integration_type not in INTEGRATION_TYPE_TO_PROVIDER:
+                return Response(
+                    {"detail": "Invalid integration type"},
+                    status=400,
+                )
+            provider_slugs = [
+                provider.value
+                for provider in INTEGRATION_TYPE_TO_PROVIDER.get(integration_type, [])
+            ]
+            queryset = queryset.filter(integration__provider__in=provider_slugs)
+
         def on_results(results: Sequence[OrganizationIntegration]) -> Sequence[Mapping[str, Any]]:
             if feature_filters:
                 results = filter_by_features(results, feature_filters)

src/sentry/integrations/base.py

@@ -125,6 +125,56 @@ class IntegrationFeatures(Enum):
     DEPLOYMENT = "deployment"
 
 
+# Integration Types
+MESSAGING = "messaging"
+PROJECT_MANAGEMENT = "project_management"
+SOURCE_CODE_MANAGEMENT = "source_code_management"
+ON_CALL_SCHEDULING = "on_call_scheduling"
+
+
+class IntegrationProviderSlug(Enum):
+    SLACK = "slack"
+    DISCORD = "discord"
+    MSTeams = "msteams"
+    JIRA = "jira"
+    JIRA_SERVER = "jira_server"
+    AZURE_DEVOPS = "vsts"
+    GITHUB = "github"
+    GITHUB_ENTERPRISE = "github_enterprise"
+    GITLAB = "gitlab"
+    BITBUCKET = "bitbucket"
+    PAGERDUTY = "pagerduty"
+    OPSGENIE = "opsgenie"
+
+
+INTEGRATION_TYPE_TO_PROVIDER = {
+    MESSAGING: [
+        IntegrationProviderSlug.SLACK,
+        IntegrationProviderSlug.DISCORD,
+        IntegrationProviderSlug.MSTeams,
+    ],
+    PROJECT_MANAGEMENT: [
+        IntegrationProviderSlug.JIRA,
+        IntegrationProviderSlug.JIRA_SERVER,
+        IntegrationProviderSlug.GITHUB,
+        IntegrationProviderSlug.GITHUB_ENTERPRISE,
+        IntegrationProviderSlug.GITLAB,
+        IntegrationProviderSlug.AZURE_DEVOPS,
+    ],
+    SOURCE_CODE_MANAGEMENT: [
+        IntegrationProviderSlug.GITHUB,
+        IntegrationProviderSlug.GITHUB_ENTERPRISE,
+        IntegrationProviderSlug.GITLAB,
+        IntegrationProviderSlug.BITBUCKET,
+        IntegrationProviderSlug.AZURE_DEVOPS,
+    ],
+    ON_CALL_SCHEDULING: [
+        IntegrationProviderSlug.PAGERDUTY,
+        IntegrationProviderSlug.OPSGENIE,
+    ],
+}
+
+
 class IntegrationProvider(PipelineProvider, abc.ABC):
     """
     An integration provider describes a third party that can be registered within Sentry.

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

@@ -15,11 +15,29 @@ def setUp(self):
             name="Example",
             external_id="example:1",
         )
+        self.msteams_integration = self.create_integration(
+            organization=self.organization,
+            provider="msteams",
+            name="MS Teams",
+            external_id="msteams:1",
+        )
+        self.opsgenie = self.create_integration(
+            organization=self.organization,
+            provider="opsgenie",
+            name="Opsgenie",
+            external_id="opsgenie:1",
+        )
+        self.slack_integration = self.create_integration(
+            organization=self.organization,
+            provider="slack",
+            name="Slack",
+            external_id="slack:1",
+        )
 
     def test_simple(self):
         response = self.get_success_response(self.organization.slug)
 
-        assert len(response.data) == 1
+        assert len(response.data) == 4
         assert response.data[0]["id"] == str(self.integration.id)
         assert "configOrganization" in response.data[0]
 
@@ -51,3 +69,40 @@ def test_provider_key(self):
             self.organization.slug, qs_params={"provider_key": "vercel"}
         )
         assert response.data == []
+
+    def test_integration_type(self):
+        response = self.get_success_response(
+            self.organization.slug, qs_params={"integrationType": "messaging"}
+        )
+        assert len(response.data) == 2
+        assert response.data[0]["id"] == str(self.msteams_integration.id)
+        assert response.data[1]["id"] == str(self.slack_integration.id)
+        response = self.get_success_response(
+            self.organization.slug, qs_params={"integrationType": "on_call_scheduling"}
+        )
+        assert len(response.data) == 1
+        assert response.data[0]["id"] == str(self.opsgenie.id)
+        response = self.get_error_response(
+            self.organization.slug, qs_params={"integrationType": "third_party"}
+        )
+        assert response.data == {"detail": "Invalid integration type"}
+        assert response.status_code == 400
+
+    def test_provider_key_and_integration_type(self):
+        response = self.get_success_response(
+            self.organization.slug,
+            qs_params={"providerKey": "slack", "integrationType": "messaging"},
+        )
+        assert len(response.data) == 1
+        assert response.data[0]["id"] == str(self.slack_integration.id)
+        response = self.get_success_response(
+            self.organization.slug,
+            qs_params={"providerKey": "vercel", "integrationType": "messaging"},
+        )
+        assert response.data == []
+        response = self.get_error_response(
+            self.organization.slug,
+            qs_params={"providerKey": "slack", "integrationType": "third_party"},
+        )
+        assert response.data == {"detail": "Invalid integration type"}
+        assert response.status_code == 400