feat(aci): add alertrule/workflow GET endpoint

src/sentry/workflow_engine/endpoints/organization_alertrule_workflow_index.py

@@ -0,0 +1,73 @@
+from drf_spectacular.utils import extend_schema
+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 region_silo_endpoint
+from sentry.api.bases import OrganizationEndpoint
+from sentry.api.bases.organization import OrganizationDetectorPermission
+from sentry.api.exceptions import ResourceDoesNotExist
+from sentry.api.serializers import serialize
+from sentry.apidocs.constants import (
+    RESPONSE_BAD_REQUEST,
+    RESPONSE_FORBIDDEN,
+    RESPONSE_NOT_FOUND,
+    RESPONSE_UNAUTHORIZED,
+)
+from sentry.apidocs.parameters import GlobalParams
+from sentry.workflow_engine.endpoints.serializers.alertrule_workflow_serializer import (
+    AlertRuleWorkflowSerializer,
+)
+from sentry.workflow_engine.endpoints.validators.alertrule_workflow import (
+    AlertRuleWorkflowValidator,
+)
+from sentry.workflow_engine.models.alertrule_workflow import AlertRuleWorkflow
+
+
+@region_silo_endpoint
+class OrganizationAlertRuleWorkflowIndexEndpoint(OrganizationEndpoint):
+    publish_status = {
+        "GET": ApiPublishStatus.EXPERIMENTAL,
+    }
+    owner = ApiOwner.ISSUES
+    permission_classes = (OrganizationDetectorPermission,)
+
+    @extend_schema(
+        operation_id="Fetch Dual-Written Rule/Alert Rules and Workflows",
+        parameters=[
+            GlobalParams.ORG_ID_OR_SLUG,
+        ],
+        responses={
+            200: AlertRuleWorkflowSerializer,
+            400: RESPONSE_BAD_REQUEST,
+            401: RESPONSE_UNAUTHORIZED,
+            403: RESPONSE_FORBIDDEN,
+            404: RESPONSE_NOT_FOUND,
+        },
+    )
+    def get(self, request, organization):
+        """
+        Returns a dual-written rule/alert rule and its associated workflow.
+        """
+        validator = AlertRuleWorkflowValidator(data=request.query_params)
+        validator.is_valid(raise_exception=True)
+        rule_id = validator.validated_data.get("rule_id")
+        alert_rule_id = validator.validated_data.get("alert_rule_id")
+        workflow_id = validator.validated_data.get("workflow_id")
+
+        queryset = AlertRuleWorkflow.objects.filter(workflow__organization=organization)
+
+        if workflow_id:
+            queryset = queryset.filter(workflow_id=workflow_id)
+
+        if alert_rule_id:
+            queryset = queryset.filter(alert_rule_id=alert_rule_id)
+
+        if rule_id:
+            queryset = queryset.filter(rule_id=rule_id)
+
+        alert_rule_workflow = queryset.first()
+        if not alert_rule_workflow:
+            raise ResourceDoesNotExist
+
+        return Response(serialize(alert_rule_workflow, request.user))

src/sentry/workflow_engine/endpoints/serializers/alertrule_workflow_serializer.py

@@ -0,0 +1,23 @@
+from collections.abc import Mapping
+from typing import Any, TypedDict
+
+from sentry.api.serializers import Serializer, register
+from sentry.workflow_engine.models import AlertRuleWorkflow
+
+
+class ActionHandlerSerializerResponse(TypedDict):
+    ruleId: str | None
+    alertRuleId: str | None
+    workflowId: str
+
+
+@register(AlertRuleWorkflow)
+class AlertRuleWorkflowSerializer(Serializer):
+    def serialize(
+        self, obj: AlertRuleWorkflow, attrs: Mapping[str, Any], user, **kwargs
+    ) -> ActionHandlerSerializerResponse:
+        return {
+            "ruleId": str(obj.rule_id) if obj.rule_id else None,
+            "alertRuleId": str(obj.alert_rule_id) if obj.alert_rule_id else None,
+            "workflowId": str(obj.workflow.id),
+        }

src/sentry/workflow_engine/endpoints/urls.py

@@ -1,5 +1,9 @@
 from django.urls import re_path
 
+from sentry.workflow_engine.endpoints.organization_alertrule_workflow_index import (
+    OrganizationAlertRuleWorkflowIndexEndpoint,
+)
+
 from .organization_available_action_index import OrganizationAvailableActionIndexEndpoint
 from .organization_data_condition_index import OrganizationDataConditionIndexEndpoint
 from .organization_detector_count import OrganizationDetectorCountEndpoint
@@ -97,4 +101,9 @@
         OrganizationOpenPeriodsEndpoint.as_view(),
         name="sentry-api-0-organization-open-periods",
     ),
+    re_path(
+        r"^(?P<organization_id_or_slug>[^/]+)/alert-rule-workflow/$",
+        OrganizationAlertRuleWorkflowIndexEndpoint.as_view(),
+        name="sentry-api-0-organization-alert-rule-workflow-index",
+    ),
 ]

src/sentry/workflow_engine/endpoints/validators/alertrule_workflow.py

@@ -0,0 +1,19 @@
+from rest_framework import serializers
+
+
+class AlertRuleWorkflowValidator(serializers.Serializer):
+    rule_id = serializers.CharField(required=False)
+    alert_rule_id = serializers.CharField(required=False)
+    workflow_id = serializers.CharField(required=False)
+
+    def validate(self, attrs):
+        super().validate(attrs)
+        if (
+            not attrs.get("rule_id")
+            and not attrs.get("alert_rule_id")
+            and not attrs.get("workflow_id")
+        ):
+            raise serializers.ValidationError(
+                "One of 'rule_id', 'alert_rule_id', or 'workflow_id' must be provided."
+            )
+        return attrs

static/app/utils/api/knownSentryApiUrls.generated.ts

@@ -121,10 +121,8 @@ export type KnownSentryApiUrls =
   | '/internal/prevent/pr-review/configs/resolved/'
   | '/internal/prevent/pr-review/github/sentry-org/'
   | '/internal/project-config/'
-  | '/internal/queue/tasks/'
   | '/internal/rpc/$serviceName/$methodName/'
   | '/internal/seer-rpc/$methodName/'
-  | '/internal/stats/'
   | '/internal/warnings/'
   | '/issues/$issueId/'
   | '/issues/$issueId/activities/'
@@ -192,6 +190,7 @@ export type KnownSentryApiUrls =
   | '/organizations/$organizationIdOrSlug/access-requests/'
   | '/organizations/$organizationIdOrSlug/access-requests/$requestId/'
   | '/organizations/$organizationIdOrSlug/ai-conversations/'
+  | '/organizations/$organizationIdOrSlug/alert-rule-workflow/'
   | '/organizations/$organizationIdOrSlug/alert-rules/'
   | '/organizations/$organizationIdOrSlug/alert-rules/$alertRuleId/'
   | '/organizations/$organizationIdOrSlug/alert-rules/available-actions/'

tests/sentry/workflow_engine/endpoints/test_organization_alertrule_workflow.py

@@ -0,0 +1,85 @@
+from sentry.api.serializers import serialize
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.silo import region_silo_test
+
+
+class OrganizationAlertRuleWorkflowAPITestCase(APITestCase):
+    endpoint = "sentry-api-0-organization-alert-rule-workflow-index"
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.login_as(user=self.user)
+
+        self.workflow_1 = self.create_workflow(organization=self.organization)
+        self.workflow_2 = self.create_workflow(organization=self.organization)
+        self.workflow_3 = self.create_workflow(organization=self.organization)
+
+        self.alert_rule_workflow_1 = self.create_alert_rule_workflow(
+            alert_rule_id=12345, workflow=self.workflow_1
+        )
+        self.alert_rule_workflow_2 = self.create_alert_rule_workflow(
+            rule_id=67890, workflow=self.workflow_2
+        )
+        self.alert_rule_workflow_3 = self.create_alert_rule_workflow(
+            alert_rule_id=11111, workflow=self.workflow_3
+        )
+
+        # Create workflow in different organization to test filtering
+        self.other_org = self.create_organization()
+        self.other_workflow = self.create_workflow(organization=self.other_org)
+        self.other_alert_rule_workflow = self.create_alert_rule_workflow(
+            alert_rule_id=99999, workflow=self.other_workflow
+        )
+
+
+@region_silo_test
+class OrganizationAlertRuleWorkflowIndexGetTest(OrganizationAlertRuleWorkflowAPITestCase):
+    def test_get_with_workflow_id_filter(self) -> None:
+        response = self.get_success_response(
+            self.organization.slug, workflow_id=str(self.workflow_1.id)
+        )
+        assert response.data == serialize(self.alert_rule_workflow_1, self.user)
+
+    def test_get_with_alert_rule_id_filter(self) -> None:
+        response = self.get_success_response(self.organization.slug, alert_rule_id="12345")
+
+        assert response.data["alertRuleId"] == "12345"
+        assert response.data["ruleId"] is None
+        assert response.data["workflowId"] == str(self.workflow_1.id)
+
+    def test_get_with_rule_id_filter(self) -> None:
+        response = self.get_success_response(self.organization.slug, rule_id="67890")
+
+        assert response.data["ruleId"] == "67890"
+        assert response.data["alertRuleId"] is None
+        assert response.data["workflowId"] == str(self.workflow_2.id)
+
+    def test_get_with_multiple_filters(self) -> None:
+        response = self.get_success_response(
+            self.organization.slug,
+            workflow_id=str(self.workflow_1.id),
+            alert_rule_id="12345",
+        )
+
+        assert response.data == serialize(self.alert_rule_workflow_1, self.user)
+
+    def test_get_with_multiple_filters_with_invalid_filter(self) -> None:
+        self.get_error_response(
+            self.organization.slug,
+            workflow_id=str(self.workflow_1.id),
+            alert_rule_id="this is not a valid ID",
+        )
+
+    def test_get_with_nonexistent_workflow_id(self) -> None:
+        self.get_error_response(self.organization.slug, workflow_id="99999", status_code=404)
+
+    def test_get_with_nonexistent_alert_rule_id(self) -> None:
+        self.get_error_response(self.organization.slug, alert_rule_id="99999", status_code=404)
+
+    def test_get_with_nonexistent_rule_id(self) -> None:
+        self.get_error_response(self.organization.slug, rule_id="99999", status_code=404)
+
+    def test_organization_isolation(self) -> None:
+        self.get_error_response(
+            self.organization.slug, workflow_id=str(self.other_workflow.id), status_code=404
+        )