feat(aci): add AlertRuleDetector lookup endpoint

src/sentry/workflow_engine/endpoints/organization_alertrule_detector_index.py

@@ -0,0 +1,75 @@
+from drf_spectacular.utils import extend_schema
+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 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.models.organization import Organization
+from sentry.workflow_engine.endpoints.serializers.alertrule_detector_serializer import (
+    AlertRuleDetectorSerializer,
+)
+from sentry.workflow_engine.endpoints.validators.alertrule_detector import (
+    AlertRuleDetectorValidator,
+)
+from sentry.workflow_engine.models.alertrule_detector import AlertRuleDetector
+
+
+@region_silo_endpoint
+class OrganizationAlertRuleDetectorIndexEndpoint(OrganizationEndpoint):
+    publish_status = {
+        "GET": ApiPublishStatus.EXPERIMENTAL,
+    }
+    owner = ApiOwner.ISSUES
+    permission_classes = (OrganizationDetectorPermission,)
+
+    @extend_schema(
+        operation_id="Fetch Dual-Written Rule/Alert Rules and Detectors",
+        parameters=[
+            GlobalParams.ORG_ID_OR_SLUG,
+        ],
+        responses={
+            200: AlertRuleDetectorSerializer,
+            400: RESPONSE_BAD_REQUEST,
+            401: RESPONSE_UNAUTHORIZED,
+            403: RESPONSE_FORBIDDEN,
+            404: RESPONSE_NOT_FOUND,
+        },
+    )
+    def get(self, request: Request, organization: Organization) -> Response:
+        """
+        Returns a dual-written rule/alert rule and its associated detector.
+        """
+        validator = AlertRuleDetectorValidator(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")
+        detector_id = validator.validated_data.get("detector_id")
+
+        queryset = AlertRuleDetector.objects.filter(detector__project__organization=organization)
+
+        if detector_id:
+            queryset = queryset.filter(detector_id=detector_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_detector = queryset.first()
+        if not alert_rule_detector:
+            raise ResourceDoesNotExist
+
+        return Response(serialize(alert_rule_detector, request.user))

src/sentry/workflow_engine/endpoints/serializers/alertrule_detector_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 AlertRuleDetector
+
+
+class AlertRuleDetectorSerializerResponse(TypedDict):
+    ruleId: str | None
+    alertRuleId: str | None
+    detectorId: str
+
+
+@register(AlertRuleDetector)
+class AlertRuleDetectorSerializer(Serializer):
+    def serialize(
+        self, obj: AlertRuleDetector, attrs: Mapping[str, Any], user, **kwargs
+    ) -> AlertRuleDetectorSerializerResponse:
+        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,
+            "detectorId": str(obj.detector.id),
+        }

src/sentry/workflow_engine/endpoints/urls.py

@@ -1,5 +1,8 @@
 from django.urls import re_path
 
+from sentry.workflow_engine.endpoints.organization_alertrule_detector_index import (
+    OrganizationAlertRuleDetectorIndexEndpoint,
+)
 from sentry.workflow_engine.endpoints.organization_alertrule_workflow_index import (
     OrganizationAlertRuleWorkflowIndexEndpoint,
 )
@@ -106,4 +109,9 @@
         OrganizationAlertRuleWorkflowIndexEndpoint.as_view(),
         name="sentry-api-0-organization-alert-rule-workflow-index",
     ),
+    re_path(
+        r"^(?P<organization_id_or_slug>[^/]+)/alert-rule-detector/$",
+        OrganizationAlertRuleDetectorIndexEndpoint.as_view(),
+        name="sentry-api-0-organization-alert-rule-detector-index",
+    ),
 ]

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

@@ -0,0 +1,19 @@
+from rest_framework import serializers
+
+
+class AlertRuleDetectorValidator(serializers.Serializer):
+    rule_id = serializers.CharField(required=False)
+    alert_rule_id = serializers.CharField(required=False)
+    detector_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("detector_id")
+        ):
+            raise serializers.ValidationError(
+                "One of 'rule_id', 'alert_rule_id', or 'detector_id' must be provided."
+            )
+        return attrs

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

@@ -190,6 +190,7 @@ export type KnownSentryApiUrls =
   | '/organizations/$organizationIdOrSlug/access-requests/'
   | '/organizations/$organizationIdOrSlug/access-requests/$requestId/'
   | '/organizations/$organizationIdOrSlug/ai-conversations/'
+  | '/organizations/$organizationIdOrSlug/alert-rule-detector/'
   | '/organizations/$organizationIdOrSlug/alert-rule-workflow/'
   | '/organizations/$organizationIdOrSlug/alert-rules/'
   | '/organizations/$organizationIdOrSlug/alert-rules/$alertRuleId/'
@@ -553,6 +554,7 @@ export type KnownSentryApiUrls =
   | '/organizations/$organizationIdOrSlug/trace-items/attributes/'
   | '/organizations/$organizationIdOrSlug/trace-items/attributes/$key/values/'
   | '/organizations/$organizationIdOrSlug/trace-items/attributes/ranked/'
+  | '/organizations/$organizationIdOrSlug/trace-items/stats/'
   | '/organizations/$organizationIdOrSlug/trace-logs/'
   | '/organizations/$organizationIdOrSlug/trace-meta/$traceId/'
   | '/organizations/$organizationIdOrSlug/trace-summary/'

tests/sentry/workflow_engine/endpoints/test_organization_alertrule_detector.py

@@ -0,0 +1,91 @@
+from sentry.api.serializers import serialize
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.silo import region_silo_test
+
+
+class OrganizationAlertRuleDetectorAPITestCase(APITestCase):
+    endpoint = "sentry-api-0-organization-alert-rule-detector-index"
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.login_as(user=self.user)
+
+        self.project = self.create_project(organization=self.organization)
+
+        self.detector_1 = self.create_detector(project=self.project)
+        self.detector_2 = self.create_detector(project=self.project)
+        self.detector_3 = self.create_detector(project=self.project)
+
+        self.alert_rule_detector_1 = self.create_alert_rule_detector(
+            alert_rule_id=12345, detector=self.detector_1
+        )
+        self.alert_rule_detector_2 = self.create_alert_rule_detector(
+            rule_id=67890, detector=self.detector_2
+        )
+        self.alert_rule_detector_3 = self.create_alert_rule_detector(
+            alert_rule_id=11111, detector=self.detector_3
+        )
+
+        # Create detector in different organization to test filtering
+        self.other_org = self.create_organization()
+        self.other_project = self.create_project(organization=self.other_org)
+        self.other_detector = self.create_detector(project=self.other_project)
+        self.other_alert_rule_detector = self.create_alert_rule_detector(
+            alert_rule_id=99999, detector=self.other_detector
+        )
+
+
+@region_silo_test
+class OrganizationAlertRuleDetectorIndexGetTest(OrganizationAlertRuleDetectorAPITestCase):
+    def test_get_with_detector_id_filter(self) -> None:
+        response = self.get_success_response(
+            self.organization.slug, detector_id=str(self.detector_1.id)
+        )
+        assert response.data == serialize(self.alert_rule_detector_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["detectorId"] == str(self.detector_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["detectorId"] == str(self.detector_2.id)
+
+    def test_get_with_multiple_filters(self) -> None:
+        response = self.get_success_response(
+            self.organization.slug,
+            detector_id=str(self.detector_1.id),
+            alert_rule_id="12345",
+        )
+
+        assert response.data == serialize(self.alert_rule_detector_1, self.user)
+
+    def test_get_with_multiple_filters_with_invalid_filter(self) -> None:
+        self.get_error_response(
+            self.organization.slug,
+            detector_id=str(self.detector_1.id),
+            alert_rule_id="this is not a valid ID",
+        )
+
+    def test_get_with_nonexistent_detector_id(self) -> None:
+        self.get_error_response(self.organization.slug, detector_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, detector_id=str(self.other_detector.id), status_code=404
+        )
+
+    def test_get_without_any_filter(self) -> None:
+        self.get_error_response(self.organization.slug, status_code=400)