ref(rules): Fix typing for rule_snooze & rule

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

@@ -15,6 +15,7 @@
 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.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
@@ -76,14 +77,15 @@ def serialize(self, obj, attrs, user, **kwargs):
 @region_silo_endpoint
 class BaseRuleSnoozeEndpoint(ProjectEndpoint):
     permission_classes = (ProjectAlertRulePermission,)
-    rule_model = type[Rule] | type[AlertRule]
+    rule_model: type[AlertRule | Rule]
     rule_field = Literal["rule", "alert_rule"]
 
     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: BaseQuerySet[AlertRule | Rule]
+            if issubclass(self.rule_model, AlertRule):
                 queryset = self.rule_model.objects.fetch_for_project(project)
             else:
                 queryset = self.rule_model.objects.filter(project=project)
@@ -104,20 +106,25 @@ def post(self, request: Request, project: Project, rule: Rule | AlertRule) -> Re
 
         if not can_edit_alert_rule(project.organization, request):
             raise PermissionDenied(
-                detail="Requesting user cannot mute this rule.", code=status.HTTP_403_FORBIDDEN
+                detail="Requesting user cannot mute this rule.", code=str(status.HTTP_403_FORBIDDEN)
             )
 
-        kwargs = {self.rule_field: rule}
-
         user_id = request.user.id if data.get("target") == "me" else None
+
+        # Due to mypy requiring keyword arguments to not be kwargs, we need to check the type of rule and query accordingly
+        # In RuleSnooze, rule and alert_rule are mutually exclusive (cannot both be not null)
+        rule_parameter = rule if isinstance(rule, Rule) else None
+        alert_rule_parameter = None if isinstance(rule, Rule) else rule
+
         rule_snooze, created = RuleSnooze.objects.get_or_create(
             user_id=user_id,
+            rule=rule_parameter,
+            alert_rule=alert_rule_parameter,
             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:
@@ -158,9 +165,14 @@ def delete(self, request: Request, project: Project, rule: Rule | AlertRule) ->
         # 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)
+            # Due to mypy requiring keyword arguments to not be kwargs, we need to check the type of rule and query accordingly
+            # In RuleSnooze, rule and alert_rule are mutually exclusive (cannot both be not null)
+            rule_parameter = rule if isinstance(rule, Rule) else None
+            alert_rule_parameter = None if isinstance(rule, Rule) else rule
+            shared_snooze = RuleSnooze.objects.get(
+                rule=rule_parameter, alert_rule=alert_rule_parameter, user_id=None
+            )
         except RuleSnooze.DoesNotExist:
             pass
 
@@ -169,11 +181,13 @@ def delete(self, request: Request, project: Project, rule: Rule | AlertRule) ->
             shared_snooze.delete()
             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)
+            rule_parameter = rule if isinstance(rule, Rule) else None
+            alert_rule_parameter = None if isinstance(rule, Rule) else rule
+            my_snooze = RuleSnooze.objects.get(
+                rule=rule_parameter, alert_rule=alert_rule_parameter, user_id=request.user.id
+            )
         except RuleSnooze.DoesNotExist:
             pass
         else:
@@ -197,7 +211,8 @@ 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.",
+                code=str(status.HTTP_403_FORBIDDEN),
             )
         # no snooze at all found
         return Response(
@@ -224,5 +239,8 @@ class MetricRuleSnoozeEndpoint(BaseRuleSnoozeEndpoint):
         "DELETE": ApiPublishStatus.UNKNOWN,
         "POST": ApiPublishStatus.UNKNOWN,
     }
-    rule_model = AlertRule
     rule_field = "alert_rule"
+    rule_model = AlertRule
+
+    def test(self):
+        self.rule_model

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

@@ -5,14 +5,15 @@
 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.models.rule import Rule
 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 +47,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 +69,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: str = [
+                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,7 +178,9 @@ def validate_conditions(self, conditions):
     def validate(self, attrs):
         return super().validate(validate_actions(attrs))
 
-    def save(self, rule):
+    def save(self, **kwargs: Any) -> Rule:
+        rule = kwargs.get("rule")
+        assert isinstance(rule, Rule), "Rule must exist to modify instance"
         rule.project = self.context["project"]
         if "environment" in self.validated_data:
             environment = self.validated_data["environment"]
@@ -187,7 +199,7 @@ def save(self, rule):
             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 = actor
             rule.owner_user_id = actor.user_id
             rule.owner_team_id = actor.team_id
         rule.save()