ref(types): Add types to rules (#31590)

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

mypy.ini

@@ -59,9 +59,7 @@ files = src/sentry/analytics/,
         src/sentry/ratelimits/,
         src/sentry/release_health/,
         src/sentry/roles/manager.py,
-        src/sentry/rules/conditions/,
-        src/sentry/rules/filters/,
-        src/sentry/rules/history/,
+        src/sentry/rules/,
         src/sentry/search/base.py,
         src/sentry/search/events/builder.py,
         src/sentry/search/events/constants.py,

src/sentry/integrations/jira/notify_action.py

@@ -56,4 +56,4 @@ class JiraCreateTicketAction(TicketEventAction):
     @transaction_start("JiraCreateTicketAction.after")
     def after(self, event, state):
         self.fix_data_for_issue()
-        yield super().after(event, state)
+        yield super().after(event, state)  # type: ignore

src/sentry/integrations/slack/notify_action.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Optional, Sequence, Tuple
+from typing import Any, Generator, Sequence
 
 from django import forms
 from django.core.exceptions import ValidationError
@@ -11,8 +11,9 @@ from sentry.eventstore.models import Event
 from sentry.integrations.slack.message_builder.issues import build_group_attachment
 from sentry.models import Integration
 from sentry.notifications.additional_attachment_manager import get_additional_attachment
+from sentry.rules import EventState, RuleFuture
 from sentry.rules.actions.base import IntegrationEventAction
-from sentry.rules.processor import RuleFuture
+from sentry.rules.base import CallbackFuture
 from sentry.shared_integrations.exceptions import (
     ApiError,
     ApiRateLimitedError,
@@ -81,7 +82,7 @@ class SlackNotifyServiceForm(forms.Form):  # type: ignore
 
         cleaned_data: dict[str, Any] = super().clean()
 
-        workspace: Optional[int] = cleaned_data.get("workspace")
+        workspace: int | None = cleaned_data.get("workspace")
 
         if channel_id:
             try:
@@ -161,7 +162,7 @@ class SlackNotifyServiceForm(forms.Form):  # type: ignore
         return cleaned_data
 
 
-class SlackNotifyServiceAction(IntegrationEventAction):  # type: ignore
+class SlackNotifyServiceAction(IntegrationEventAction):
     id = "sentry.integrations.slack.notify_action.SlackNotifyServiceAction"
     form_cls = SlackNotifyServiceForm
     label = "Send a notification to the {workspace} Slack workspace to {channel} (optionally, an ID: {channel_id}) and show tags {tags} in notification"
@@ -181,7 +182,7 @@ class SlackNotifyServiceAction(IntegrationEventAction):  # type: ignore
             "tags": {"type": "string", "placeholder": "i.e environment,user,my_tag"},
         }
 
-    def after(self, event: Event, state: str) -> Any:
+    def after(self, event: Event, state: EventState) -> Generator[CallbackFuture, None, None]:
         channel = self.get_option("channel_id")
         tags = set(self.get_tags_list())
 
@@ -245,7 +246,5 @@ class SlackNotifyServiceAction(IntegrationEventAction):  # type: ignore
             self.data, integrations=self.get_integrations(), channel_transformer=self.get_channel_id
         )
 
-    def get_channel_id(
-        self, integration: Integration, name: str
-    ) -> Tuple[str, Optional[str], bool]:
+    def get_channel_id(self, integration: Integration, name: str) -> tuple[str, str | None, bool]:
         return get_channel_id(self.project.organization, integration, name)

src/sentry/integrations/vsts/notify_action.py

@@ -1,15 +1,17 @@
 import logging
-from typing import Any
+from typing import Generator
 
 from sentry.eventstore.models import Event
+from sentry.rules import EventState
 from sentry.rules.actions.base import TicketEventAction
+from sentry.rules.base import CallbackFuture
 from sentry.utils.http import absolute_uri
 from sentry.web.decorators import transaction_start
 
 logger = logging.getLogger("sentry.rules")
 
 
-class AzureDevopsCreateTicketAction(TicketEventAction):  # type: ignore
+class AzureDevopsCreateTicketAction(TicketEventAction):
     id = "sentry.integrations.vsts.notify_action.AzureDevopsCreateTicketAction"
     label = "Create an Azure DevOps work item in {integration} with these "
     ticket_type = "an Azure DevOps work item"
@@ -19,10 +21,11 @@ class AzureDevopsCreateTicketAction(TicketEventAction):  # type: ignore
 
     def generate_footer(self, rule_url: str) -> str:
         return "\nThis work item was automatically created by Sentry via [{}]({})".format(
-            self.rule.label,
+            # TODO(mgaeta): Bug: Rule is optional.
+            self.rule.label,  # type: ignore
             absolute_uri(rule_url),
         )
 
     @transaction_start("AzureDevopsCreateTicketAction.after")
-    def after(self, event: Event, state: str) -> Any:
-        yield super().after(event, state)
+    def after(self, event: Event, state: EventState) -> Generator[CallbackFuture, None, None]:
+        yield super().after(event, state)  # type: ignore

src/sentry/mail/adapter.py

@@ -12,6 +12,7 @@ from sentry.notifications.notifications.rules import AlertRuleNotification
 from sentry.notifications.notifications.user_report import UserReportNotification
 from sentry.notifications.types import ActionTargetType
 from sentry.plugins.base.structs import Notification
+from sentry.rules import RuleFuture
 from sentry.tasks.digests import deliver_digest
 from sentry.utils import metrics
 
@@ -29,7 +30,7 @@ class MailAdapter:
     def rule_notify(
         self,
         event: Any,
-        futures: Sequence[Any],
+        futures: Sequence[RuleFuture],
         target_type: ActionTargetType,
         target_identifier: Optional[int] = None,
     ) -> None:

src/sentry/rules/__init__.py

@@ -1,3 +1,7 @@
+from collections import namedtuple
+
+RuleFuture = namedtuple("RuleFuture", ["rule", "kwargs"])
+
 from .base import EventState, RuleBase
 from .match import LEVEL_MATCH_CHOICES, MATCH_CHOICES, MatchType
 from .registry import RuleRegistry
@@ -9,6 +13,7 @@ __all__ = (
     "MATCH_CHOICES",
     "MatchType",
     "RuleBase",
+    "RuleFuture",
     "rules",
 )
 

src/sentry/rules/actions/base.py

@@ -2,22 +2,28 @@ from __future__ import annotations
 
 import abc
 import logging
+from typing import Any, Callable, Generator, Mapping, Sequence
 
 from django import forms
+from django.db.models import QuerySet
+from rest_framework.response import Response
 
 from sentry.constants import ObjectStatus
+from sentry.eventstore.models import Event
+from sentry.integrations import IntegrationInstallation
 from sentry.models import ExternalIssue, GroupLink, Integration
-from sentry.rules.base import RuleBase
+from sentry.rules import RuleFuture
+from sentry.rules.base import CallbackFuture, EventState, RuleBase
 
 logger = logging.getLogger("sentry.rules")
 
 INTEGRATION_KEY = "integration"
 
 
-class IntegrationNotifyServiceForm(forms.Form):
+class IntegrationNotifyServiceForm(forms.Form):  # type: ignore
     integration = forms.ChoiceField(choices=(), widget=forms.Select())
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         integrations = [(i.id, i.name) for i in kwargs.pop("integrations")]
         super().__init__(*args, **kwargs)
         if integrations:
@@ -31,7 +37,7 @@ class EventAction(RuleBase, abc.ABC):
     rule_type = "action/event"
 
     @abc.abstractmethod
-    def after(self, event, state):
+    def after(self, event: Event, state: EventState) -> Generator[CallbackFuture, None, None]:
         """
         Executed after a Rule matches.
 
@@ -72,29 +78,33 @@ class IntegrationEventAction(EventAction, abc.ABC):
     def integration_key(self) -> str:
         pass
 
-    def is_enabled(self):
-        return self.get_integrations().exists()
+    def is_enabled(self) -> bool:
+        enabled: bool = self.get_integrations().exists()
+        return enabled
 
-    def get_integration_name(self):
-        """
-        Get the integration's name for the label.
-
-        :return: string
-        """
+    def get_integration_name(self) -> str:
+        """Get the integration's name for the label."""
         try:
-            return self.get_integration().name
+            integration = self.get_integration()
         except Integration.DoesNotExist:
             return "[removed]"
 
-    def get_integrations(self):
-        return Integration.objects.get_active_integrations(self.project.organization.id).filter(
+        _name: str = integration.name
+        return _name
+
+    def get_integrations(self) -> QuerySet[Integration]:
+        query: QuerySet[Integration] = Integration.objects.get_active_integrations(
+            self.project.organization.id
+        ).filter(
             provider=self.provider,
         )
+        return query
 
-    def get_integration_id(self):
-        return self.get_option(self.integration_key)
+    def get_integration_id(self) -> str:
+        integration_id: str = self.get_option(self.integration_key)
+        return integration_id
 
-    def get_integration(self):
+    def get_integration(self) -> Integration:
         """
         Uses the required class variables `provider` and `integration_key` with
         RuleBase.get_option to get the integration object from DB.
@@ -107,14 +117,19 @@ class IntegrationEventAction(EventAction, abc.ABC):
             provider=self.provider,
         )
 
-    def get_installation(self):
+    def get_installation(self) -> Any:
         return self.get_integration().get_installation(self.project.organization.id)
 
-    def get_form_instance(self):
+    def get_form_instance(self) -> forms.Form:
         return self.form_cls(self.data, integrations=self.get_integrations())
 
 
-def create_link(integration, installation, event, response):
+def create_link(
+    integration: Integration,
+    installation: IntegrationInstallation,
+    event: Event,
+    response: Response,
+) -> None:
     """
     After creating the event on a third-party service, create a link to the
     external resource in the DB. TODO make this a transaction.
@@ -143,17 +158,25 @@ def create_link(integration, installation, event, response):
     )
 
 
-def build_description(event, rule_id, installation, generate_footer):
+def build_description(
+    event: Event,
+    rule_id: int,
+    installation: IntegrationInstallation,
+    generate_footer: Callable[[str], str],
+) -> str:
     """
     Format the description of the ticket/work item
     """
     project = event.group.project
     rule_url = f"/organizations/{project.organization.slug}/alerts/rules/{project.slug}/{rule_id}/"
 
-    return installation.get_group_description(event.group, event) + generate_footer(rule_url)
+    description: str = installation.get_group_description(event.group, event) + generate_footer(
+        rule_url
+    )
+    return description
 
 
-def create_issue(event, futures):
+def create_issue(event: Event, futures: Sequence[RuleFuture]) -> None:
     """Create an issue for a given event"""
     organization = event.group.project.organization
 
@@ -201,7 +224,7 @@ class TicketEventAction(IntegrationEventAction, abc.ABC):
 
     form_cls = IntegrationNotifyServiceForm
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         super(IntegrationEventAction, self).__init__(*args, **kwargs)
         integration_choices = [
             (i.id, self.translate_integration(i)) for i in self.get_integrations()
@@ -224,8 +247,9 @@ class TicketEventAction(IntegrationEventAction, abc.ABC):
         if dynamic_fields:
             self.form_fields.update(dynamic_fields)
 
-    def render_label(self):
-        return self.label.format(integration=self.get_integration_name())
+    def render_label(self) -> str:
+        label: str = self.label.format(integration=self.get_integration_name())
+        return label
 
     @property
     @abc.abstractmethod
@@ -236,13 +260,13 @@ class TicketEventAction(IntegrationEventAction, abc.ABC):
     def prompt(self) -> str:
         return f"Create {self.ticket_type}"
 
-    def get_dynamic_form_fields(self):
+    def get_dynamic_form_fields(self) -> Mapping[str, Any] | None:
         """
         Either get the dynamic form fields cached on the DB return `None`.
 
         :return: (Option) Django form fields dictionary
         """
-        form_fields = self.data.get("dynamic_form_fields")
+        form_fields: Mapping[str, Any] | list[Any] | None = self.data.get("dynamic_form_fields")
         if not form_fields:
             return None
 
@@ -255,17 +279,19 @@ class TicketEventAction(IntegrationEventAction, abc.ABC):
             return fields
         return form_fields
 
-    def translate_integration(self, integration):
-        return integration.name
+    def translate_integration(self, integration: Integration) -> str:
+        name: str = integration.name
+        return name
 
     @abc.abstractmethod
-    def generate_footer(self, rule_url):
+    def generate_footer(self, rule_url: str) -> str:
         pass
 
-    def after(self, event, state):
+    def after(self, event: Event, state: EventState) -> Generator[CallbackFuture, None, None]:
         integration_id = self.get_integration_id()
         key = f"{self.provider}:{integration_id}"
-        return self.future(
+        # TODO(mgaeta): Bug: Inheriting functions all _yield_ not return.
+        return self.future(  # type: ignore
             create_issue,
             key=key,
             data=self.data,

src/sentry/rules/actions/notify_event.py

@@ -1,20 +1,23 @@
-"""
-Used for notifying *all* enabled plugins
-"""
+from typing import Generator, Sequence
 
+from sentry.eventstore.models import Event
 from sentry.plugins.base import plugins
+from sentry.rules import EventState
 from sentry.rules.actions.base import EventAction
 from sentry.rules.actions.services import LegacyPluginService
+from sentry.rules.base import CallbackFuture
 from sentry.utils import metrics
 from sentry.utils.safe import safe_execute
 
 
 class NotifyEventAction(EventAction):
+    """Used for notifying *all* enabled plugins."""
+
     id = "sentry.rules.actions.notify_event.NotifyEventAction"
     label = "Send a notification (for all legacy integrations)"
     prompt = "Send a notification to all legacy integrations"
 
-    def get_plugins(self):
+    def get_plugins(self) -> Sequence[LegacyPluginService]:
         from sentry.plugins.bases.notify import NotificationPlugin
 
         results = []
@@ -29,12 +32,12 @@ class NotifyEventAction(EventAction):
 
         return results
 
-    def after(self, event, state):
+    def after(self, event: Event, state: EventState) -> Generator[CallbackFuture, None, None]:
         group = event.group
 
-        for plugin in self.get_plugins():
+        for plugin_ in self.get_plugins():
             # plugin is now wrapped in the LegacyPluginService object
-            plugin = plugin.service
+            plugin = plugin_.service
             if not safe_execute(
                 plugin.should_notify, group=group, event=event, _with_transaction=False
             ):

src/sentry/rules/actions/notify_event_sentry_app.py

@@ -1,7 +1,6 @@
-"""
-Used for notifying a *specific* sentry app with a custom webhook payload (i.e. specified UI components)
-"""
-from typing import Any, Mapping, Optional, Sequence
+from __future__ import annotations
+
+from typing import Any, Generator, Mapping, Sequence
 
 from rest_framework import serializers
 
@@ -9,16 +8,18 @@ from sentry.api.serializers import serialize
 from sentry.api.serializers.models.sentry_app_component import SentryAppAlertRuleActionSerializer
 from sentry.eventstore.models import Event
 from sentry.models import Project, SentryApp, SentryAppComponent, SentryAppInstallation
+from sentry.rules import EventState
 from sentry.rules.actions.base import EventAction
+from sentry.rules.base import CallbackFuture
 from sentry.tasks.sentry_apps import notify_sentry_app
 
 ValidationError = serializers.ValidationError
 
 
-def validate_field(value: Optional[str], field: Mapping[str, Any], app_name: str):
+def validate_field(value: str | None, field: Mapping[str, Any], app_name: str) -> None:
     # Only validate synchronous select fields
     if field.get("type") == "select" and not field.get("uri"):
-        allowed_values = [option[0] for option in field.get("options")]
+        allowed_values = [option[0] for option in field.get("options", [])]
         # Reject None values and empty strings
         if value and value not in allowed_values:
             field_label = field.get("label")
@@ -28,7 +29,12 @@ def validate_field(value: Optional[str], field: Mapping[str, Any], app_name: str
             )
 
 
-class NotifyEventSentryAppAction(EventAction):  # type: ignore
+class NotifyEventSentryAppAction(EventAction):
+    """
+    Used for notifying a *specific* sentry app with a custom webhook payload
+    (i.e. specified UI components).
+    """
+
     id = "sentry.rules.actions.notify_event_sentry_app.NotifyEventSentryAppAction"
     actionType = "sentryapp"
     # Required field for EventAction, value is ignored
@@ -52,7 +58,7 @@ class NotifyEventSentryAppAction(EventAction):  # type: ignore
 
         return action_list
 
-    def get_sentry_app(self, event: Event) -> Optional[SentryApp]:
+    def get_sentry_app(self, event: Event) -> SentryApp | None:
         extra = {"event_id": event.event_id}
 
         sentry_app_installation_uuid = self.get_option("sentryAppInstallationUuid")
@@ -67,7 +73,7 @@ class NotifyEventSentryAppAction(EventAction):  # type: ignore
 
         return None
 
-    def get_setting_value(self, field_name):
+    def get_setting_value(self, field_name: str) -> str | None:
         incoming_settings = self.data.get("settings", [])
         return next(
             (setting["value"] for setting in incoming_settings if setting["name"] == field_name),
@@ -133,7 +139,7 @@ class NotifyEventSentryAppAction(EventAction):  # type: ignore
                 f"Unexpected setting(s) '{extra_keys_string}' configured for {sentry_app.name}"
             )
 
-    def after(self, event: Event, state: str) -> Any:
+    def after(self, event: Event, state: EventState) -> Generator[CallbackFuture, None, None]:
         sentry_app = self.get_sentry_app(event)
         yield self.future(
             notify_sentry_app,

src/sentry/rules/actions/notify_event_service.py

@@ -1,7 +1,7 @@
-"""
-Used for notifying a *specific* plugin/sentry app with a generic webhook payload
-"""
+from __future__ import annotations
+
 import logging
+from typing import Any, Generator, Mapping, Sequence
 
 from django import forms
 
@@ -10,12 +10,20 @@ from sentry.api.serializers import serialize
 from sentry.api.serializers.models.app_platform_event import AppPlatformEvent
 from sentry.api.serializers.models.incident import IncidentSerializer
 from sentry.constants import SentryAppInstallationStatus
-from sentry.incidents.models import INCIDENT_STATUS, IncidentStatus
+from sentry.eventstore.models import Event
+from sentry.incidents.models import (
+    INCIDENT_STATUS,
+    AlertRuleTriggerAction,
+    Incident,
+    IncidentStatus,
+)
 from sentry.integrations.metric_alerts import incident_attachment_info
 from sentry.models import SentryApp, SentryAppInstallation
 from sentry.plugins.base import plugins
+from sentry.rules import EventState
 from sentry.rules.actions.base import EventAction
 from sentry.rules.actions.services import PluginService, SentryAppService
+from sentry.rules.base import CallbackFuture
 from sentry.tasks.sentry_apps import notify_sentry_app, send_and_save_webhook_request
 from sentry.utils import metrics
 from sentry.utils.safe import safe_execute
@@ -24,7 +32,11 @@ logger = logging.getLogger("sentry.integrations.sentry_app")
 PLUGINS_WITH_FIRST_PARTY_EQUIVALENTS = ["PagerDuty", "Slack"]
 
 
-def build_incident_attachment(incident, new_status: IncidentStatus, metric_value=None):
+def build_incident_attachment(
+    incident: Incident,
+    new_status: IncidentStatus,
+    metric_value: str | None = None,
+) -> Mapping[str, str]:
     from sentry.api.serializers.rest_framework.base import (
         camel_to_snake_case,
         convert_dict_key_case,
@@ -42,8 +54,11 @@ def build_incident_attachment(incident, new_status: IncidentStatus, metric_value
 
 
 def send_incident_alert_notification(
-    action, incident, new_status: IncidentStatus, metric_value=None
-):
+    action: AlertRuleTriggerAction,
+    incident: Incident,
+    new_status: IncidentStatus,
+    metric_value: str | None = None,
+) -> None:
     """
     When a metric alert is triggered, send incident data to the SentryApp's webhook.
     :param action: The triggered `AlertRuleTriggerAction`.
@@ -72,7 +87,7 @@ def send_incident_alert_notification(
                 "sentry_app_id": sentry_app.id,
             },
         )
-        return
+        return None
 
     app_platform_event = AppPlatformEvent(
         resource="metric_alert",
@@ -121,10 +136,10 @@ def find_alert_rule_action_ui_component(app_platform_event: AppPlatformEvent) ->
     return bool(len(actions))
 
 
-class NotifyEventServiceForm(forms.Form):
+class NotifyEventServiceForm(forms.Form):  # type: ignore
     service = forms.ChoiceField(choices=())
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         service_choices = [(s.slug, s.title) for s in kwargs.pop("services")]
 
         super().__init__(*args, **kwargs)
@@ -134,12 +149,14 @@ class NotifyEventServiceForm(forms.Form):
 
 
 class NotifyEventServiceAction(EventAction):
+    """Used for notifying a *specific* plugin/sentry app with a generic webhook payload."""
+
     id = "sentry.rules.actions.notify_event_service.NotifyEventServiceAction"
     form_cls = NotifyEventServiceForm
     label = "Send a notification via {service}"
     prompt = "Send a notification via an integration"
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self.form_fields = {
             "service": {
@@ -148,12 +165,12 @@ class NotifyEventServiceAction(EventAction):
             }
         }
 
-    def transform_title(self, title):
+    def transform_title(self, title: str) -> str:
         if title in PLUGINS_WITH_FIRST_PARTY_EQUIVALENTS:
             return f"(Legacy) {title}"
         return title
 
-    def after(self, event, state):
+    def after(self, event: Event, state: EventState) -> Generator[CallbackFuture, None, None]:
         service = self.get_option("service")
 
         extra = {"event_id": event.event_id}
@@ -199,7 +216,7 @@ class NotifyEventServiceAction(EventAction):
             metrics.incr("notifications.sent", instance=plugin.slug, skip_internal=False)
             yield self.future(plugin.rule_notify)
 
-    def get_sentry_app_services(self):
+    def get_sentry_app_services(self) -> Sequence[SentryAppService]:
         # excludes Sentry Apps that have Alert Rule UI Component in their schema
         return [
             SentryAppService(app)
@@ -207,7 +224,7 @@ class NotifyEventServiceAction(EventAction):
             if not SentryAppService(app).has_alert_rule_action()
         ]
 
-    def get_plugins(self):
+    def get_plugins(self) -> Sequence[PluginService]:
         from sentry.plugins.bases.notify import NotificationPlugin
 
         results = []
@@ -222,10 +239,8 @@ class NotifyEventServiceAction(EventAction):
 
         return results
 
-    def get_services(self):
-        services = self.get_plugins()
-        services += self.get_sentry_app_services()
-        return services
+    def get_services(self) -> Sequence[Any]:
+        return [*self.get_plugins(), *self.get_sentry_app_services()]
 
-    def get_form_instance(self):
+    def get_form_instance(self) -> forms.Form:
         return self.form_cls(self.data, services=self.get_services())