ref(rules): Fix typing for rule_snooze & rule (#80306)

Author: Christinarlong

pyproject.toml

@@ -157,7 +157,6 @@ module = [
     "sentry.api.endpoints.project_rules_configuration",
     "sentry.api.endpoints.project_servicehook_stats",
     "sentry.api.endpoints.project_transaction_names",
-    "sentry.api.endpoints.rule_snooze",
     "sentry.api.endpoints.team_details",
     "sentry.api.endpoints.team_release_count",
     "sentry.api.endpoints.user_subscriptions",
@@ -182,7 +181,6 @@ module = [
     "sentry.api.serializers.models.team",
     "sentry.api.serializers.rest_framework.mentions",
     "sentry.api.serializers.rest_framework.notification_action",
-    "sentry.api.serializers.rest_framework.rule",
     "sentry.auth.helper",
     "sentry.auth.provider",
     "sentry.auth.system",

src/sentry/api/endpoints/rule_snooze.py

@@ -1,7 +1,8 @@
 import datetime
-from typing import Literal
+from typing import Any, Generic, TypeVar
 
 from django.contrib.auth.models import AnonymousUser
+from django.core.exceptions import ObjectDoesNotExist
 from rest_framework import serializers, status
 from rest_framework.exceptions import NotFound, PermissionDenied
 from rest_framework.request import Request
@@ -15,6 +16,8 @@
 from sentry.api.exceptions import BadRequest
 from sentry.api.serializers import Serializer, register, serialize
 from sentry.api.serializers.rest_framework.base import CamelSnakeSerializer
+from sentry.db.models.base import Model
+from sentry.db.models.manager.base_query_set import BaseQuerySet
 from sentry.incidents.models.alert_rule import AlertRule
 from sentry.models.organization import Organization
 from sentry.models.organizationmember import OrganizationMember
@@ -73,69 +76,61 @@ def serialize(self, obj, attrs, user, **kwargs):
         return result
 
 
-@region_silo_endpoint
-class BaseRuleSnoozeEndpoint(ProjectEndpoint):
+T = TypeVar("T", bound=Model)
+
+
+class BaseRuleSnoozeEndpoint(ProjectEndpoint, Generic[T]):
     permission_classes = (ProjectAlertRulePermission,)
-    rule_model = type[Rule] | type[AlertRule]
-    rule_field = Literal["rule", "alert_rule"]
+    rule_field: str  # abstract, value comes from child class
 
     def convert_args(self, request: Request, rule_id: int, *args, **kwargs):
         (args, kwargs) = super().convert_args(request, *args, **kwargs)
         project = kwargs["project"]
         try:
-            if self.rule_model is AlertRule:
-                queryset = self.rule_model.objects.fetch_for_project(project)
-            else:
-                queryset = self.rule_model.objects.filter(project=project)
+            queryset = self.fetch_rule_list(project=project)
             rule = queryset.get(id=rule_id)
-        except self.rule_model.DoesNotExist:
+        except ObjectDoesNotExist:
             raise NotFound(detail="Rule does not exist")
 
         kwargs["rule"] = rule
 
         return (args, kwargs)
 
-    def post(self, request: Request, project: Project, rule: Rule | AlertRule) -> Response:
+    def post(self, request: Request, project: Project, rule: T) -> Response:
         serializer = RuleSnoozeValidator(data=request.data)
         if not serializer.is_valid():
             return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
 
         data = serializer.validated_data
 
         if not can_edit_alert_rule(project.organization, request):
-            raise PermissionDenied(
-                detail="Requesting user cannot mute this rule.", code=status.HTTP_403_FORBIDDEN
-            )
-
-        kwargs = {self.rule_field: rule}
+            raise PermissionDenied(detail="Requesting user cannot mute this rule.")
 
         user_id = request.user.id if data.get("target") == "me" else None
-        rule_snooze, created = RuleSnooze.objects.get_or_create(
-            user_id=user_id,
-            defaults={
-                "owner_id": request.user.id,
-                "until": data.get("until"),
-                "date_added": datetime.datetime.now(datetime.UTC),
-            },
-            **kwargs,
-        )
-        # don't allow editing of a rulesnooze object for a given rule and user (or no user)
-        if not created:
+
+        try:
+            rule_snooze = self.fetch_instance(
+                user_id=user_id,
+                rule=rule,
+            )
+            # don't allow editing of a rulesnooze object for a given rule and user (or no user)
             return Response(
                 {"detail": "RuleSnooze already exists for this rule and scope."},
                 status=status.HTTP_410_GONE,
             )
+        except RuleSnooze.DoesNotExist:
+            rule_snooze = self.create_instance(
+                rule=rule,
+                user_id=user_id,
+                owner_id=request.user.id,
+                until=data.get("until"),
+                date_added=datetime.datetime.now(datetime.UTC),
+            )
 
         if not user_id:
             # create an audit log entry if the rule is snoozed for everyone
-            audit_log_event = "RULE_SNOOZE" if self.rule_model == Rule else "ALERT_RULE_SNOOZE"
-
-            self.create_audit_entry(
-                request=request,
-                organization=project.organization,
-                target_object=rule.id,
-                event=audit_log.get_event_id(audit_log_event),
-                data=rule.get_audit_log_data(),
+            self.record_audit_log_entry(
+                request=request, organization=project.organization, rule=rule
             )
 
         analytics.record(
@@ -154,13 +149,12 @@ def post(self, request: Request, project: Project, rule: Rule | AlertRule) -> Re
             status=status.HTTP_201_CREATED,
         )
 
-    def delete(self, request: Request, project: Project, rule: Rule | AlertRule) -> Response:
+    def delete(self, request: Request, project: Project, rule: T) -> Response:
         # find if there is a mute for all that I can remove
         shared_snooze = None
         deletion_type = None
-        kwargs = {self.rule_field: rule, "user_id": None}
         try:
-            shared_snooze = RuleSnooze.objects.get(**kwargs)
+            shared_snooze = self.fetch_instance(user_id=None, rule=rule)
         except RuleSnooze.DoesNotExist:
             pass
 
@@ -170,10 +164,9 @@ def delete(self, request: Request, project: Project, rule: Rule | AlertRule) ->
             deletion_type = "everyone"
 
         # next check if there is a mute for me that I can remove
-        kwargs = {self.rule_field: rule, "user_id": request.user.id}
         my_snooze = None
         try:
-            my_snooze = RuleSnooze.objects.get(**kwargs)
+            my_snooze = self.fetch_instance(user_id=request.user.id, rule=rule)
         except RuleSnooze.DoesNotExist:
             pass
         else:
@@ -197,32 +190,98 @@ def delete(self, request: Request, project: Project, rule: Rule | AlertRule) ->
         # didn't find a match but there is a shared snooze
         if shared_snooze:
             raise PermissionDenied(
-                detail="Requesting user cannot unmute this rule.", code=status.HTTP_403_FORBIDDEN
+                detail="Requesting user cannot unmute this rule.",
             )
         # no snooze at all found
         return Response(
             {"detail": "This rulesnooze object doesn't exist."},
             status=status.HTTP_404_NOT_FOUND,
         )
 
+    def record_audit_log_entry(
+        self, request: Request, organization: Organization, rule: T, **kwargs: Any
+    ) -> None:
+        raise NotImplementedError()
+
+    def fetch_instance(self, rule: T, user_id: int | None, **kwargs: Any) -> RuleSnooze:
+        raise NotImplementedError()
+
+    def create_instance(self, rule: T, user_id: int | None, **kwargs: Any) -> RuleSnooze:
+        raise NotImplementedError()
+
+    def fetch_rule_list(self, project: Project) -> BaseQuerySet[T]:
+        raise NotImplementedError()
+
 
 @region_silo_endpoint
-class RuleSnoozeEndpoint(BaseRuleSnoozeEndpoint):
+class RuleSnoozeEndpoint(BaseRuleSnoozeEndpoint[Rule]):
     owner = ApiOwner.ISSUES
     publish_status = {
         "DELETE": ApiPublishStatus.UNKNOWN,
         "POST": ApiPublishStatus.UNKNOWN,
     }
-    rule_model = Rule
     rule_field = "rule"
 
+    def fetch_rule_list(self, project: Project) -> BaseQuerySet[Rule]:
+        queryset = Rule.objects.filter(project=project)
+
+        return queryset
+
+    def fetch_instance(self, rule: Rule, user_id: int | None, **kwargs: Any) -> RuleSnooze:
+        rule_snooze = RuleSnooze.objects.get(user_id=user_id, rule=rule, **kwargs)
+
+        return rule_snooze
+
+    def create_instance(self, rule: Rule, user_id: int | None, **kwargs: Any) -> RuleSnooze:
+        rule_snooze = RuleSnooze.objects.create(user_id=user_id, rule=rule, **kwargs)
+
+        return rule_snooze
+
+    def record_audit_log_entry(
+        self, request: Request, organization: Organization, rule: Rule, **kwargs: Any
+    ) -> None:
+        self.create_audit_entry(
+            request=request,
+            organization=organization,
+            target_object=rule.id,
+            event=audit_log.get_event_id("RULE_SNOOZE"),
+            data=rule.get_audit_log_data(),
+            **kwargs,
+        )
+
 
 @region_silo_endpoint
-class MetricRuleSnoozeEndpoint(BaseRuleSnoozeEndpoint):
+class MetricRuleSnoozeEndpoint(BaseRuleSnoozeEndpoint[AlertRule]):
     owner = ApiOwner.ISSUES
     publish_status = {
         "DELETE": ApiPublishStatus.UNKNOWN,
         "POST": ApiPublishStatus.UNKNOWN,
     }
-    rule_model = AlertRule
     rule_field = "alert_rule"
+
+    def fetch_rule_list(self, project: Project) -> BaseQuerySet[AlertRule]:
+        queryset = AlertRule.objects.fetch_for_project(project=project)
+
+        return queryset
+
+    def fetch_instance(self, rule: AlertRule, user_id: int | None, **kwargs: Any) -> RuleSnooze:
+        rule_snooze = RuleSnooze.objects.get(user_id=user_id, alert_rule=rule, **kwargs)
+
+        return rule_snooze
+
+    def create_instance(self, rule: AlertRule, user_id: int | None, **kwargs: Any) -> RuleSnooze:
+        rule_snooze = RuleSnooze.objects.create(user_id=user_id, alert_rule=rule, **kwargs)
+
+        return rule_snooze
+
+    def record_audit_log_entry(
+        self, request: Request, organization: Organization, rule: AlertRule, **kwargs: Any
+    ) -> None:
+        self.create_audit_entry(
+            request=request,
+            organization=organization,
+            target_object=rule.id,
+            event=audit_log.get_event_id("ALERT_RULE_SNOOZE"),
+            data=rule.get_audit_log_data(),
+            **kwargs,
+        )

src/sentry/api/serializers/rest_framework/rule.py

@@ -5,14 +5,14 @@
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
+from rest_framework.exceptions import ValidationError
 
 from sentry import features
 from sentry.api.fields.actor import ActorField
 from sentry.constants import MIGRATED_CONDITIONS, SENTRY_APP_ACTIONS, TICKET_ACTIONS
 from sentry.models.environment import Environment
 from sentry.rules import rules
-
-ValidationError = serializers.ValidationError
+from sentry.rules.actions.sentry_apps.notify_event import NotifyEventSentryAppAction
 
 
 @extend_schema_field(field=OpenApiTypes.OBJECT)
@@ -46,12 +46,16 @@ def to_internal_value(self, data):
             msg = "Invalid node. Could not find '%s'"
             raise ValidationError(msg % data["id"])
 
+        # instance of a RuleBase class
         node = cls(project=self.context["project"], data=data)
 
         # Nodes with user-declared fields will manage their own validation
         if node.id in SENTRY_APP_ACTIONS:
             if not data.get("hasSchemaFormConfig"):
                 raise ValidationError("Please configure your integration settings.")
+            assert isinstance(
+                node, NotifyEventSentryAppAction
+            ), "node must be an instance of NotifyEventSentryAppAction to use self_validate"
             node.self_validate()
             return data
 
@@ -64,10 +68,15 @@ def to_internal_value(self, data):
             # XXX(epurkhiser): Very hacky, but we really just want validation
             # errors that are more specific, not just 'this wasn't filled out',
             # give a more generic error for those.
-            first_error = next(iter(form.errors.values()))[0]
+            # Still hacky, but a bit clearer, for each field there can be a list of errors so
+            # we get the first error's message for each field and then get the first error message from that list
+            first_error_message = [
+                error_list[0]["message"]
+                for field, error_list in form.errors.get_json_data().items()
+            ][0]
 
-            if first_error != "This field is required.":
-                raise ValidationError(first_error)
+            if first_error_message != "This field is required.":
+                raise ValidationError(first_error_message)
 
             raise ValidationError("Ensure all required fields are filled in.")
 
@@ -168,31 +177,6 @@ def validate_conditions(self, conditions):
     def validate(self, attrs):
         return super().validate(validate_actions(attrs))
 
-    def save(self, rule):
-        rule.project = self.context["project"]
-        if "environment" in self.validated_data:
-            environment = self.validated_data["environment"]
-            rule.environment_id = int(environment) if environment else environment
-        if self.validated_data.get("name"):
-            rule.label = self.validated_data["name"]
-        if self.validated_data.get("actionMatch"):
-            rule.data["action_match"] = self.validated_data["actionMatch"]
-        if self.validated_data.get("filterMatch"):
-            rule.data["filter_match"] = self.validated_data["filterMatch"]
-        if self.validated_data.get("actions") is not None:
-            rule.data["actions"] = self.validated_data["actions"]
-        if self.validated_data.get("conditions") is not None:
-            rule.data["conditions"] = self.validated_data["conditions"]
-        if self.validated_data.get("frequency"):
-            rule.data["frequency"] = self.validated_data["frequency"]
-        if self.validated_data.get("owner"):
-            actor = self.validated_data["owner"].resolve_to_actor()
-            rule.owner_id = actor.id
-            rule.owner_user_id = actor.user_id
-            rule.owner_team_id = actor.team_id
-        rule.save()
-        return rule
-
 
 ACTION_UUID_KEY = "uuid"