ref: fix typing for rules.actions.integrations.base (#84785)

rather than specifying form_cls / get_form_instance separately -- just
define get_form_instance

<!-- Describe your PR here. -->

Author: asottile-sentry

pyproject.toml

@@ -215,7 +215,6 @@ module = [
     "sentry.plugins.config",
     "sentry.receivers.releases",
     "sentry.release_health.metrics_sessions_v2",
-    "sentry.rules.actions.integrations.base",
     "sentry.rules.actions.integrations.create_ticket.form",
     "sentry.rules.actions.integrations.create_ticket.utils",
     "sentry.rules.history.preview",

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

@@ -59,10 +59,9 @@ def to_internal_value(self, data):
             node.self_validate()
             return data
 
-        if not node.form_cls:
-            return data
-
         form = node.get_form_instance()
+        if not form:
+            return data
 
         if not form.is_valid():
             # XXX(epurkhiser): Very hacky, but we really just want validation

src/sentry/integrations/discord/actions/issue_alert/notification.py

@@ -20,7 +20,6 @@
 
 class DiscordNotifyServiceAction(IntegrationEventAction):
     id = "sentry.integrations.discord.notify_action.DiscordNotifyServiceAction"
-    form_cls = DiscordNotifyServiceForm
     label = "Send a notification to the {server} Discord server in the channel with ID or URL: {channel_id} and show tags {tags} in the notification."
     prompt = "Send a Discord notification"
     provider = "discord"
@@ -95,5 +94,5 @@ def render_label(self) -> str:
     def get_tags_list(self) -> Sequence[str]:
         return [s.strip() for s in self.get_option("tags", "").split(",")]
 
-    def get_form_instance(self) -> Any:
-        return self.form_cls(self.data, integrations=self.get_integrations())
+    def get_form_instance(self) -> DiscordNotifyServiceForm:
+        return DiscordNotifyServiceForm(self.data, integrations=self.get_integrations())

src/sentry/integrations/jira/actions/create_ticket.py

@@ -14,7 +14,6 @@ class JiraCreateTicketAction(TicketEventAction):
     ticket_type = "a Jira issue"
     link = "https://docs.sentry.io/product/integrations/issue-tracking/jira/#issue-sync"
     provider = "jira"
-    form_cls = JiraNotifyServiceForm
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
@@ -32,3 +31,6 @@ def generate_footer(self, rule_url: str) -> str:
     def translate_integration(self, integration: RpcIntegration) -> str:
         name = integration.metadata.get("domain_name", integration.name)
         return name.replace(".atlassian.net", "")
+
+    def get_form_instance(self) -> JiraNotifyServiceForm:
+        return JiraNotifyServiceForm(self.data, integrations=self.get_integrations())

src/sentry/integrations/jira_server/actions/create_ticket.py

@@ -14,7 +14,6 @@ class JiraServerCreateTicketAction(TicketEventAction):
     ticket_type = "a Jira Server issue"
     link = "https://docs.sentry.io/product/integrations/issue-tracking/jira/#issue-sync"
     provider = "jira_server"
-    form_cls = JiraServerNotifyServiceForm
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
@@ -31,3 +30,6 @@ def generate_footer(self, rule_url: str) -> str:
 
     def translate_integration(self, integration: RpcIntegration) -> str:
         return integration.metadata.get("domain_name", integration.name)
+
+    def get_form_instance(self) -> JiraServerNotifyServiceForm:
+        return JiraServerNotifyServiceForm(self.data, integrations=self.get_integrations())

src/sentry/integrations/msteams/actions/notification.py

@@ -19,7 +19,6 @@
 
 class MsTeamsNotifyServiceAction(IntegrationEventAction):
     id = "sentry.integrations.msteams.notify_action.MsTeamsNotifyServiceAction"
-    form_cls = MsTeamsNotifyServiceForm
     label = "Send a notification to the {team} Team to {channel}"
     prompt = "Send a Microsoft Teams notification"
     provider = "msteams"
@@ -78,8 +77,8 @@ def render_label(self):
             team=self.get_integration_name(), channel=self.get_option("channel")
         )
 
-    def get_form_instance(self):
-        return self.form_cls(
+    def get_form_instance(self) -> MsTeamsNotifyServiceForm:
+        return MsTeamsNotifyServiceForm(
             self.data, integrations=self.get_integrations(), channel_transformer=self.get_channel_id
         )
 

src/sentry/integrations/opsgenie/actions/notification.py

@@ -17,7 +17,6 @@
 
 class OpsgenieNotifyTeamAction(IntegrationEventAction):
     id = "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction"
-    form_cls = OpsgenieNotifyTeamForm
     label = (
         "Send a notification to Opsgenie account {account} and team {team} with {priority} priority"
     )
@@ -127,8 +126,8 @@ def render_label(self) -> str:
             account=self.get_integration_name(), team=team_name, priority=priority
         )
 
-    def get_form_instance(self):
-        return self.form_cls(
+    def get_form_instance(self) -> OpsgenieNotifyTeamForm:
+        return OpsgenieNotifyTeamForm(
             self.data,
             org_id=self.project.organization_id,
             integrations=self.get_integrations(),

src/sentry/integrations/pagerduty/actions/notification.py

@@ -16,7 +16,6 @@
 
 class PagerDutyNotifyServiceAction(IntegrationEventAction):
     id = "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction"
-    form_cls = PagerDutyNotifyServiceForm
     label = "Send a notification to PagerDuty account {account} and service {service} with {severity} severity"
     prompt = "Send a PagerDuty notification"
     provider = "pagerduty"
@@ -141,8 +140,8 @@ def render_label(self):
             account=self.get_integration_name(), service=service_name, severity=severity
         )
 
-    def get_form_instance(self):
-        return self.form_cls(
+    def get_form_instance(self) -> PagerDutyNotifyServiceForm:
+        return PagerDutyNotifyServiceForm(
             self.data,
             integrations=self.get_integrations(),
             services=self.get_services(),

src/sentry/integrations/slack/actions/notification.py

@@ -53,7 +53,6 @@
 
 class SlackNotifyServiceAction(IntegrationEventAction):
     id = "sentry.integrations.slack.notify_action.SlackNotifyServiceAction"
-    form_cls = SlackNotifyServiceForm
     prompt = "Send a Slack notification"
     provider = "slack"
     integration_key = "workspace"
@@ -335,8 +334,8 @@ def render_label(self) -> str:
     def get_tags_list(self) -> Sequence[str]:
         return [s.strip() for s in self.get_option("tags", "").split(",")]
 
-    def get_form_instance(self) -> Any:
-        return self.form_cls(
+    def get_form_instance(self) -> SlackNotifyServiceForm:
+        return SlackNotifyServiceForm(
             self.data, integrations=self.get_integrations(), channel_transformer=self.get_channel_id
         )
 

src/sentry/mail/actions.py

@@ -17,7 +17,6 @@
 
 class NotifyEmailAction(EventAction):
     id = "sentry.mail.actions.NotifyEmailAction"
-    form_cls = NotifyEmailForm
     label = "Send a notification to {targetType} and if none can be found then send a notification to {fallthroughType}"
     prompt = "Send a notification"
     metrics_slug = "EmailAction"
@@ -73,5 +72,5 @@ def after(self, event, notification_uuid: str | None = None):
             )
         )
 
-    def get_form_instance(self):
-        return self.form_cls(self.project, self.data)
+    def get_form_instance(self) -> NotifyEmailForm:
+        return NotifyEmailForm(self.project, self.data)

src/sentry/rules/actions/integrations/base.py

@@ -2,8 +2,6 @@
 
 import abc
 
-from django import forms
-
 from sentry import analytics
 from sentry.eventstore.models import GroupEvent
 from sentry.integrations.services.integration import (
@@ -83,9 +81,6 @@ def get_organization_integration(self) -> RpcOrganizationIntegration | None:
             integration_id=self.get_integration_id(), organization_id=self.project.organization_id
         )
 
-    def get_form_instance(self) -> forms.Form:
-        return self.form_cls(self.data, integrations=self.get_integrations())
-
     def record_notification_sent(
         self,
         event: GroupEvent,

src/sentry/rules/actions/integrations/create_ticket/base.py

@@ -16,7 +16,6 @@
 class TicketEventAction(IntegrationEventAction, abc.ABC):
     """Shared ticket actions"""
 
-    form_cls = IntegrationNotifyServiceForm
     integration_key = "integration"
     link: str | None
     rule: Rule
@@ -97,3 +96,6 @@ def after(
             integration_id=integration_id,
             provider=self.provider,
         )
+
+    def get_form_instance(self) -> IntegrationNotifyServiceForm:
+        return IntegrationNotifyServiceForm(self.data, integrations=self.get_integrations())

src/sentry/rules/actions/notify_event_service.py

@@ -119,7 +119,6 @@ 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"
 
@@ -197,5 +196,5 @@ def get_plugins(self) -> Sequence[PluginService]:
     def get_services(self) -> Sequence[Any]:
         return [*self.get_plugins(), *self.get_sentry_app_services()]
 
-    def get_form_instance(self) -> forms.Form:
-        return self.form_cls(self.data, services=self.get_services())
+    def get_form_instance(self) -> NotifyEventServiceForm:
+        return NotifyEventServiceForm(self.data, services=self.get_services())

src/sentry/rules/base.py

@@ -55,8 +55,6 @@
 
 
 class RuleBase(abc.ABC):
-    form_cls: type[forms.Form] = None  # type: ignore[assignment]
-
     logger = logging.getLogger("sentry.rules")
 
     def __init__(
@@ -82,18 +80,18 @@ def is_enabled(self) -> bool:
     def get_option(self, key: str, default: str | None = None) -> Any:
         return self.data.get(key, default)
 
-    def get_form_instance(self) -> forms.Form:
-        return self.form_cls(self.data if self.had_data else None)
+    def get_form_instance(self) -> forms.Form | None:
+        return None
 
     def render_label(self) -> str:
         return self.label.format(**self.data)
 
     def validate_form(self) -> bool:
-        if not self.form_cls:
+        form = self.get_form_instance()
+        if form is None:
             return True
-
-        is_valid: bool = self.get_form_instance().is_valid()
-        return is_valid
+        else:
+            return form.is_valid()
 
     def future(
         self,

src/sentry/rules/conditions/event_attribute.py

@@ -90,7 +90,6 @@ class EventAttributeCondition(EventCondition):
     """
 
     id = "sentry.rules.conditions.event_attribute.EventAttributeCondition"
-    form_cls = EventAttributeForm
     label = "The event's {attribute} value {match} {value}"
 
     form_fields = {
@@ -189,6 +188,9 @@ def get_event_columns(self) -> dict[Dataset, Sequence[str]]:
         columns: dict[Dataset, Sequence[str]] = get_dataset_columns([column])
         return columns
 
+    def get_form_instance(self) -> EventAttributeForm:
+        return EventAttributeForm(self.data)
+
 
 # Register attribute handlers
 @attribute_registry.register("platform")

src/sentry/rules/conditions/event_frequency.py

@@ -131,7 +131,6 @@ class _QSTypedDict(TypedDict):
 
 class BaseEventFrequencyCondition(EventCondition, abc.ABC):
     intervals = STANDARD_INTERVALS
-    form_cls = EventFrequencyForm
 
     def __init__(
         self,
@@ -409,6 +408,9 @@ def get_value_from_groups(
             result = group.get(value)
         return result
 
+    def get_form_instance(self) -> EventFrequencyForm:
+        return EventFrequencyForm(self.data)
+
 
 class EventFrequencyCondition(BaseEventFrequencyCondition):
     id = "sentry.rules.conditions.event_frequency.EventFrequencyCondition"
@@ -820,7 +822,6 @@ class EventFrequencyPercentCondition(BaseEventFrequencyCondition):
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.intervals = PERCENT_INTERVALS
-        self.form_cls = EventFrequencyPercentForm
         super().__init__(*args, **kwargs)
 
         # Override form fields interval to hide 1 min option from ui, but leave
@@ -948,6 +949,9 @@ def passes_activity_frequency(
     ) -> bool:
         raise NotImplementedError
 
+    def get_form_instance(self) -> EventFrequencyPercentForm:
+        return EventFrequencyPercentForm(self.data)
+
 
 def bucket_count(start: datetime, end: datetime, buckets: dict[datetime, int]) -> int:
     rounded_end = round_to_five_minute(end)

src/sentry/rules/conditions/level.py

@@ -26,7 +26,6 @@ class LevelEventForm(forms.Form):
 
 class LevelCondition(EventCondition):
     id = "sentry.rules.conditions.level.LevelCondition"
-    form_cls = LevelEventForm
     label = "The event's level is {match} {level}"
     form_fields = {
         "level": {"type": "choice", "choices": list(LEVEL_CHOICES.items())},
@@ -81,3 +80,6 @@ def passes_activity(
             return self._passes(level)
         except (TypeError, KeyError):
             return False
+
+    def get_form_instance(self) -> LevelEventForm:
+        return LevelEventForm(self.data)

src/sentry/rules/conditions/tagged_event.py

@@ -36,7 +36,6 @@ def clean(self) -> dict[str, Any] | None:
 
 class TaggedEventCondition(EventCondition):
     id = "sentry.rules.conditions.tagged_event.TaggedEventCondition"
-    form_cls = TaggedEventForm
     label = "The event's tags match {key} {match} {value}"
 
     form_fields = {
@@ -113,3 +112,6 @@ def get_event_columns(self) -> dict[Dataset, Sequence[str]]:
             [Columns.TAGS_KEY, Columns.TAGS_VALUE]
         )
         return columns
+
+    def get_form_instance(self) -> TaggedEventForm:
+        return TaggedEventForm(self.data)

src/sentry/rules/filters/age_comparison.py

@@ -39,7 +39,6 @@ class AgeComparisonForm(forms.Form):
 
 class AgeComparisonFilter(EventFilter):
     id = "sentry.rules.filters.age_comparison.AgeComparisonFilter"
-    form_cls = AgeComparisonForm
     form_fields = {
         "comparison_type": {"type": "choice", "choices": age_comparison_choices},
         "value": {"type": "number", "placeholder": 10},
@@ -89,3 +88,6 @@ def passes_activity(
             return False
 
         return self._passes(group.first_seen, condition_activity.timestamp)
+
+    def get_form_instance(self) -> AgeComparisonForm:
+        return AgeComparisonForm(self.data)

src/sentry/rules/filters/assigned_to.py

@@ -20,7 +20,6 @@
 
 class AssignedToFilter(EventFilter):
     id = "sentry.rules.filters.assigned_to.AssignedToFilter"
-    form_cls = AssignedToForm
     label = "The issue is assigned to {targetType}"
     prompt = "The issue is assigned to {no one/team/member}"
 
@@ -53,8 +52,7 @@ def passes(self, event: GroupEvent, state: EventState) -> bool:
             return False
 
     def get_form_instance(self) -> forms.Form:
-        form: forms.Form = self.form_cls(self.project, self.data)
-        return form
+        return AssignedToForm(self.project, self.data)
 
     def render_label(self) -> str:
         target_type = AssigneeTargetType(self.get_option("targetType"))

src/sentry/rules/filters/issue_category.py

@@ -19,7 +19,6 @@ class IssueCategoryForm(forms.Form):
 
 class IssueCategoryFilter(EventFilter):
     id = "sentry.rules.filters.issue_category.IssueCategoryFilter"
-    form_cls = IssueCategoryForm
     form_fields = {"value": {"type": "choice", "choices": list(CATEGORY_CHOICES.items())}}
     rule_type = "filter/event"
     label = "The issue's category is equal to {value}"
@@ -54,3 +53,6 @@ def render_label(self) -> str:
         title = CATEGORY_CHOICES.get(value)
         group_category_name = title.title() if title else ""
         return self.label.format(value=group_category_name)
+
+    def get_form_instance(self) -> IssueCategoryForm:
+        return IssueCategoryForm(self.data)

src/sentry/rules/filters/issue_occurrences.py

@@ -16,7 +16,6 @@ class IssueOccurrencesForm(forms.Form):
 
 class IssueOccurrencesFilter(EventFilter):
     id = "sentry.rules.filters.issue_occurrences.IssueOccurrencesFilter"
-    form_cls = IssueOccurrencesForm
     form_fields = {"value": {"type": "number", "placeholder": 10}}
     label = "The issue has happened at least {value} times"
     prompt = "The issue has happened at least {x} times (Note: this is approximate)"
@@ -56,3 +55,6 @@ def passes_activity(
         )
 
         return bool(guess >= value)
+
+    def get_form_instance(self) -> IssueOccurrencesForm:
+        return IssueOccurrencesForm(self.data)