chore(typing): Add condition typeddict to processing (#70006)

Taken from #68517 and added as a
followup PR alongside #69068

Author: schew2381

src/sentry/api/endpoints/project_agnostic_rule_conditions.py

@@ -22,7 +22,7 @@ def get(self, request: Request, organization) -> Response:
 
         def info_extractor(rule_cls):
             context = {"id": rule_cls.id, "label": rule_cls.label}
-            node = rule_cls(None)
+            node = rule_cls(project=None)
             if hasattr(node, "form_fields"):
                 context["formFields"] = node.form_fields
 

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

@@ -20,7 +20,7 @@ def generate_rule_label(project, rule, data):
     if rule_cls is None:
         return
 
-    rule_inst = rule_cls(project, data=data, rule=rule)
+    rule_inst = rule_cls(project=project, data=data, rule=rule)
     return rule_inst.render_label()
 
 

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

@@ -46,7 +46,7 @@ def to_internal_value(self, data):
             msg = "Invalid node. Could not find '%s'"
             raise ValidationError(msg % data["id"])
 
-        node = cls(self.context["project"], data)
+        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:

src/sentry/rules/conditions/base.py

@@ -1,12 +1,19 @@
 import abc
 from collections.abc import Sequence
 from datetime import datetime
+from typing import TypedDict
 
 from sentry.eventstore.models import GroupEvent
 from sentry.rules.base import EventState, RuleBase
 from sentry.types.condition_activity import ConditionActivity
 
 
+class GenericCondition(TypedDict):
+    # the ID in the rules registry that maps to a condition class
+    # e.g. "sentry.rules.conditions.every_event.EveryEventCondition"
+    id: str
+
+
 class EventCondition(RuleBase, abc.ABC):
     rule_type = "condition/event"
 

src/sentry/rules/conditions/event_frequency.py

@@ -6,7 +6,7 @@
 from collections import defaultdict
 from collections.abc import Callable, Mapping
 from datetime import datetime, timedelta
-from typing import Any
+from typing import Any, Literal, NotRequired
 
 from django import forms
 from django.core.cache import cache
@@ -19,7 +19,7 @@
 from sentry.issues.grouptype import GroupCategory
 from sentry.models.group import Group
 from sentry.rules import EventState
-from sentry.rules.conditions.base import EventCondition
+from sentry.rules.conditions.base import EventCondition, GenericCondition
 from sentry.tsdb.base import TSDBModel
 from sentry.types.condition_activity import (
     FREQUENCY_CONDITION_BUCKET_SIZE,
@@ -55,6 +55,26 @@ class ComparisonType(TextChoices):
     PERCENT = "percent"
 
 
+class EventFrequencyConditionData(GenericCondition):
+    """
+    The base typed dict for all condition data representing EventFrequency issue
+    alert rule conditions
+    """
+
+    # Either the count or percentage.
+    value: int
+    # The interval to compare the value against such as 5m, 1h, 3w, etc.
+    # e.g. # of issues is more than {value} in {interval}.
+    interval: str
+    # NOTE: Some of tne earliest COUNT conditions were created without the
+    # comparisonType field, although modern rules will always have it.
+    comparisonType: NotRequired[Literal[ComparisonType.COUNT, ComparisonType.PERCENT]]
+    # The previous interval to compare the curr interval against. This is only
+    # present in PERCENT conditions.
+    # e.g. # of issues is 50% higher in {interval} compared to {comparisonInterval}
+    comparisonInterval: NotRequired[str]
+
+
 class EventFrequencyForm(forms.Form):
     intervals = STANDARD_INTERVALS
     interval = forms.ChoiceField(
@@ -102,6 +122,9 @@ class BaseEventFrequencyCondition(EventCondition, abc.ABC):
 
     def __init__(
         self,
+        # Data specifically takes on this typeddict form for the
+        # Event Frequency condition classes.
+        data: EventFrequencyConditionData | None = None,
         *args: Any,
         **kwargs: Any,
     ) -> None:
@@ -119,6 +142,7 @@ def __init__(
                 ],
             },
         }
+        kwargs["data"] = data
 
         super().__init__(*args, **kwargs)
 

src/sentry/rules/history/preview.py

@@ -145,7 +145,7 @@ def get_issue_state_activity(
         if condition_cls is None:
             raise PreviewException
         # instantiates a EventCondition subclass and retrieves activities related to it
-        condition_inst = condition_cls(project, data=condition)
+        condition_inst = condition_cls(project=project, data=condition)
         try:
             activities = condition_inst.get_activity(start, end, CONDITION_ACTIVITY_LIMIT)
             for activity in activities:
@@ -403,7 +403,9 @@ def apply_frequency_conditions(
         condition_cls = rules.get(condition_data["id"])
         if condition_cls is None:
             raise PreviewException
-        condition_types[condition_data["id"]].append(condition_cls(project, data=condition_data))
+        condition_types[condition_data["id"]].append(
+            condition_cls(project=project, data=condition_data)
+        )
 
     filtered_activity = defaultdict(list)
     if condition_match == "all":

src/sentry/rules/processing/delayed_processing.py

@@ -1,13 +1,16 @@
 import logging
 from collections import defaultdict
-from collections.abc import MutableMapping
-from typing import Any, DefaultDict, NamedTuple
+from typing import DefaultDict, NamedTuple
 
 from sentry.buffer.redis import BufferHookEvent, RedisBuffer, redis_buffer_registry
 from sentry.models.project import Project
 from sentry.models.rule import Rule
 from sentry.rules import rules
-from sentry.rules.conditions.event_frequency import BaseEventFrequencyCondition, ComparisonType
+from sentry.rules.conditions.event_frequency import (
+    BaseEventFrequencyCondition,
+    ComparisonType,
+    EventFrequencyConditionData,
+)
 from sentry.rules.processing.processor import is_condition_slow, split_conditions_and_filters
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
@@ -30,24 +33,24 @@ def __repr__(self):
 
 
 class DataAndGroups(NamedTuple):
-    data: MutableMapping[str, Any] | None
+    data: EventFrequencyConditionData | None
     group_ids: set[int]
 
     def __repr__(self):
         return f"data: {self.data}\ngroup_ids: {self.group_ids}"
 
 
-def get_slow_conditions(rule: Rule) -> list[MutableMapping[str, str]]:
+def get_slow_conditions(rule: Rule) -> list[EventFrequencyConditionData]:
     """
     Returns the slow conditions of a rule model instance.
     """
     conditions_and_filters = rule.data.get("conditions", ())
     conditions, _ = split_conditions_and_filters(conditions_and_filters)
-    slow_conditions: list[MutableMapping[str, str]] = [
-        cond for cond in conditions if is_condition_slow(cond)
-    ]
+    slow_conditions = [cond for cond in conditions if is_condition_slow(cond)]
 
-    return slow_conditions
+    # MyPy refuses to make TypedDict compatible with MutableMapping
+    # https://github.com/python/mypy/issues/4976
+    return slow_conditions  # type: ignore[return-value]
 
 
 def get_rules_to_groups(rulegroup_to_events: dict[str, str]) -> DefaultDict[int, set[int]]:
@@ -61,8 +64,10 @@ def get_rules_to_groups(rulegroup_to_events: dict[str, str]) -> DefaultDict[int,
 
 def get_rule_to_slow_conditions(
     alert_rules: list[Rule],
-) -> DefaultDict[Rule, list[MutableMapping[str, str]]]:
-    rule_to_slow_conditions: DefaultDict[Rule, list[MutableMapping[str, str]]] = defaultdict(list)
+) -> DefaultDict[Rule, list[EventFrequencyConditionData]]:
+    rule_to_slow_conditions: DefaultDict[Rule, list[EventFrequencyConditionData]] = defaultdict(
+        list
+    )
     for rule in alert_rules:
         slow_conditions = get_slow_conditions(rule)
         for condition_data in slow_conditions:
@@ -113,7 +118,9 @@ def get_condition_group_results(
             logger.warning("Unregistered condition %r", unique_condition.cls_id)
             return None
 
-        condition_inst = condition_cls(project=project, data=condition_data)
+        # MyPy refuses to make TypedDict compatible with MutableMapping
+        # https://github.com/python/mypy/issues/4976
+        condition_inst = condition_cls(project=project, data=condition_data)  # type: ignore[arg-type]
         if not isinstance(condition_inst, BaseEventFrequencyCondition):
             logger.warning("Unregistered condition %r", condition_cls.id)
             return None
@@ -139,7 +146,7 @@ def get_condition_group_results(
 
 def get_rules_to_fire(
     condition_group_results: dict[UniqueCondition, dict[int, int]],
-    rule_to_slow_conditions: DefaultDict[Rule, list[MutableMapping[str, str]]],
+    rule_to_slow_conditions: DefaultDict[Rule, list[EventFrequencyConditionData]],
     rules_to_groups: DefaultDict[int, set[int]],
 ) -> DefaultDict[Rule, set[int]]:
     rules_to_fire = defaultdict(set)

src/sentry/rules/processing/processor.py

@@ -64,7 +64,9 @@ def get_rule_type(condition: Mapping[str, Any]) -> str | None:
     return rule_type
 
 
-def split_conditions_and_filters(rule_condition_list):
+def split_conditions_and_filters(
+    rule_condition_list,
+) -> tuple[list[MutableMapping[str, Any]], list[MutableMapping[str, Any]]]:
     condition_list = []
     filter_list = []
     for rule_cond in rule_condition_list:

tests/sentry/rules/history/endpoints/test_project_rule_preview.py

@@ -128,7 +128,7 @@ def test_inbox_reason(self):
             frequency=10,
         )
 
-        for (group, reason) in group_reason:
+        for group, reason in group_reason:
             assert any([int(g["id"]) == group.id for g in resp.data])
 
             for preview_group in resp.data: