♻️ feat(workflow_engine): issue alert action migration util for slack

src/sentry/workflow_engine/migration_helpers/rule_action.py

@@ -1,57 +1,14 @@
+import dataclasses
 import logging
 from typing import Any
 
-from sentry.notifications.models.notificationaction import ActionTarget
 from sentry.workflow_engine.models.action import Action
 from sentry.workflow_engine.typings.notification_action import (
-    ACTION_TYPE_TO_INTEGRATION_ID_KEY,
-    ACTION_TYPE_TO_TARGET_DISPLAY_KEY,
-    ACTION_TYPE_TO_TARGET_IDENTIFIER_KEY,
-    ACTION_TYPE_TO_TARGET_TYPE_RULE_REGISTRY,
     EXCLUDED_ACTION_DATA_KEYS,
-    RULE_REGISTRY_ID_TO_INTEGRATION_PROVIDER,
-    SlackDataBlob,
+    issue_alert_action_translator_registry,
 )
 
-_logger = logging.getLogger(__name__)
-
-
-def build_slack_data_blob(action: dict[str, Any]) -> SlackDataBlob:
-    """
-    Builds a SlackDataBlob from the action data.
-    Only includes the keys that are not None.
-    """
-    return SlackDataBlob(
-        tags=action.get("tags", ""),
-        notes=action.get("notes", ""),
-    )
-
-
-def sanitize_action(action: dict[str, Any], action_type: Action.Type) -> dict[str, Any]:
-    """
-    Pops the keys we don't want to save inside the JSON field of the Action model.
-
-    :param action: action data (Rule.data.actions)
-    :param action_type: action type (Action.Type)
-    :return: action data without the excluded keys
-    """
-
-    # # If we have a specific blob type, we need to sanitize the action data to the blob type
-    if action_type == Action.Type.SLACK:
-        return build_slack_data_blob(action).__dict__
-    # # Otherwise, we can just return the action data as is, removing the keys we don't want to save
-    else:
-        return {
-            k: v
-            for k, v in action.items()
-            if k
-            not in [
-                ACTION_TYPE_TO_INTEGRATION_ID_KEY.get(action_type),
-                ACTION_TYPE_TO_TARGET_IDENTIFIER_KEY.get(action_type),
-                ACTION_TYPE_TO_TARGET_DISPLAY_KEY.get(action_type),
-                *EXCLUDED_ACTION_DATA_KEYS,
-            ]
-        }
+logger = logging.getLogger(__name__)
 
 
 def build_notification_actions_from_rule_data_actions(
@@ -62,75 +19,88 @@ def build_notification_actions_from_rule_data_actions(
 
     :param actions: list of action data (Rule.data.actions)
     :return: list of notification actions (Action)
+
+    :raises ValueError: if action is missing an id
+    :raises KeyError: if there isn't a translator registered for the id
     """
 
     notification_actions: list[Action] = []
 
     for action in actions:
-        # Use Rule.integration.provider to get the action type
-        action_type = RULE_REGISTRY_ID_TO_INTEGRATION_PROVIDER.get(action.get("id"))
-        if action_type is None:
-            _logger.warning(
-                "Action type not found for action",
+        registry_id = action.get("id")
+        if not registry_id:
+            logger.error(
+                "No registry ID found for action",
+                extra={"action_uuid": action.get("uuid")},
+            )
+            continue
+
+        try:
+            translator_class = issue_alert_action_translator_registry.get(registry_id)
+            translator = translator_class()
+        except KeyError:
+            logger.exception(
+                "Action translator not found for action",
                 extra={
-                    "action_id": action.get("id"),
+                    "registry_id": registry_id,
                     "action_uuid": action.get("uuid"),
                 },
             )
             continue
 
-        # For all integrations, the target type is specific
-        # For email, the target type is user
-        # For sentry app, the target type is sentry app
-        # FWIW, we don't use target type for issue alerts
-        target_type = ACTION_TYPE_TO_TARGET_TYPE_RULE_REGISTRY.get(action_type)
-
-        if target_type == ActionTarget.SPECIFIC:
-
-            # Get the integration_id
-            integration_id_key = ACTION_TYPE_TO_INTEGRATION_ID_KEY.get(action_type)
-            if integration_id_key is None:
-                # we should always have an integration id key if target type is specific
-                # TODO(iamrajjoshi): Should we fail loudly here?
-                _logger.warning(
-                    "Integration ID key not found for action type",
-                    extra={
-                        "action_type": action_type,
-                        "action_id": action.get("id"),
-                        "action_uuid": action.get("uuid"),
-                    },
-                )
-                continue
-            integration_id = action.get(integration_id_key)
-
-            # Get the target_identifier if it exists
-            target_identifier_key = ACTION_TYPE_TO_TARGET_IDENTIFIER_KEY.get(action_type)
-            if target_identifier_key is not None:
-                target_identifier = action.get(target_identifier_key)
-
-            # Get the target_display if it exists
-            target_display_key = ACTION_TYPE_TO_TARGET_DISPLAY_KEY.get(action_type)
-            if target_display_key is not None:
-                target_display = action.get(target_display_key)
+        # Get integration ID if needed
+        # This won't be set for all actions, (e.g. sentry app)
+        integration_id = None
+        if translator.integration_id_key:
+            integration_id = action.get(translator.integration_id_key)
+
+        # Get target identifier if needed
+        # This won't be set for all actions, (e.g. sentry app)
+        target_identifier = None
+        if translator.target_identifier_key:
+            target_identifier = action.get(translator.target_identifier_key)
+
+        # Get target display if needed
+        # This won't be set for all actions, some integrations also don't have a target display
+        target_display = None
+        if translator.target_display_key:
+            target_display = action.get(translator.target_display_key)
+
+        # Sanitize the action data
+        data = translator.sanitize_action(action)
+        if translator.blob_type:
+            # Convert to dataclass if blob type is specified
+            # k.name contains the field name inside the dataclass
+            blob_instance = translator.blob_type(
+                **{k.name: action.get(k.name, "") for k in dataclasses.fields(translator.blob_type)}
+            )
+            data = dataclasses.asdict(blob_instance)
+        else:
+            # Remove keys we don't want to save
+            data = {
+                k: v
+                for k, v in action.items()
+                if k
+                not in [
+                    translator.integration_id_key,
+                    translator.target_identifier_key,
+                    translator.target_display_key,
+                    *EXCLUDED_ACTION_DATA_KEYS,
+                ]
+            }
 
         notification_action = Action(
-            type=action_type,
-            data=(
-                # If the target type is specific, sanitize the action data
-                # Otherwise, use the action data as is
-                sanitize_action(action, action_type)
-                if target_type == ActionTarget.SPECIFIC
-                else action
-            ),
+            type=translator.action_type,
+            data=data,
             integration_id=integration_id,
             target_identifier=target_identifier,
             target_display=target_display,
-            target_type=target_type,
+            target_type=translator.target_type,
         )
 
         notification_actions.append(notification_action)
 
-    # Bulk create the actions, note: this won't call save(), not sure if we need to
+    # Bulk create the actions
     Action.objects.bulk_create(notification_actions)
 
     return notification_actions

src/sentry/workflow_engine/typings/notification_action.py

@@ -1,129 +1,90 @@
+from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from typing import Any, ClassVar
 
 from sentry.notifications.models.notificationaction import ActionTarget
+from sentry.utils.registry import Registry
 from sentry.workflow_engine.models.action import Action
 
-"""
-Keys that are excluded from the action data blob.
-We don't want to save these keys because:
-- uuid: maps to action id
-- id: maps to action type
-"""
+# Keep existing excluded keys constant
 EXCLUDED_ACTION_DATA_KEYS = ["uuid", "id"]
 
-"""
-Action types that are integrations
-"""
-INTEGRATION_ACTION_TYPES = [
-    Action.Type.SLACK,
-    Action.Type.DISCORD,
-    Action.Type.MSTEAMS,
-    Action.Type.PAGERDUTY,
-    Action.Type.OPSGENIE,
-    Action.Type.GITHUB,
-    Action.Type.GITHUB_ENTERPRISE,
-    Action.Type.JIRA,
-    Action.Type.JIRA_SERVER,
-    Action.Type.AZURE_DEVOPS,
-]
-
-ACTION_TYPE_TO_TARGET_TYPE_RULE_REGISTRY: dict[Action.Type, ActionTarget] = {
-    Action.Type.SLACK: ActionTarget.SPECIFIC,
-    Action.Type.DISCORD: ActionTarget.SPECIFIC,
-    Action.Type.MSTEAMS: ActionTarget.SPECIFIC,
-    Action.Type.PAGERDUTY: ActionTarget.SPECIFIC,
-    Action.Type.OPSGENIE: ActionTarget.SPECIFIC,
-    Action.Type.GITHUB: ActionTarget.SPECIFIC,
-    Action.Type.GITHUB_ENTERPRISE: ActionTarget.SPECIFIC,
-    Action.Type.JIRA: ActionTarget.SPECIFIC,
-    Action.Type.JIRA_SERVER: ActionTarget.SPECIFIC,
-    Action.Type.AZURE_DEVOPS: ActionTarget.SPECIFIC,
-    Action.Type.SENTRY_APP: ActionTarget.SENTRY_APP,
-    Action.Type.EMAIL: ActionTarget.USER,
-    Action.Type.PLUGIN: ActionTarget.USER,
-    Action.Type.WEBHOOK: ActionTarget.USER,
-}
-
-INTEGRATION_PROVIDER_TO_RULE_REGISTRY_ID: dict[Action.Type, str] = {
-    Action.Type.SLACK: "sentry.integrations.slack.notify_action.SlackNotifyServiceAction",
-    Action.Type.DISCORD: "sentry.integrations.discord.notify_action.DiscordNotifyServiceAction",
-    Action.Type.MSTEAMS: "sentry.integrations.msteams.notify_action.MsTeamsNotifyServiceAction",
-    Action.Type.PAGERDUTY: "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction",
-    Action.Type.OPSGENIE: "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction",
-    Action.Type.GITHUB: "sentry.integrations.github.notify_action.GitHubCreateTicketAction",
-    Action.Type.GITHUB_ENTERPRISE: "sentry.integrations.github_enterprise.notify_action.GitHubEnterpriseCreateTicketAction",
-    Action.Type.JIRA: "sentry.integrations.jira.notify_action.JiraCreateTicketAction",
-    Action.Type.JIRA_SERVER: "sentry.integrations.jira_server.notify_action.JiraServerCreateTicketAction",
-    Action.Type.AZURE_DEVOPS: "sentry.integrations.vsts.notify_action.AzureDevopsCreateTicketAction",
-    Action.Type.SENTRY_APP: "sentry.rules.actions.notify_event_sentry_app.NotifyEventSentryAppAction",
-    Action.Type.EMAIL: "sentry.mail.actions.NotifyEmailAction",
-    Action.Type.PLUGIN: "sentry.rules.actions.notify_event.NotifyEventAction",
-    Action.Type.WEBHOOK: "sentry.rules.actions.notify_event_service.NotifyEventServiceAction",
-}
-
-RULE_REGISTRY_ID_TO_INTEGRATION_PROVIDER: dict[str, Action.Type] = {
-    v: k for k, v in INTEGRATION_PROVIDER_TO_RULE_REGISTRY_ID.items()
-}
-
-ACTION_TYPE_TO_INTEGRATION_ID_KEY: dict[Action.Type, str] = {
-    Action.Type.SLACK: "workspace",
-    Action.Type.DISCORD: "server",
-    Action.Type.MSTEAMS: "team",
-    Action.Type.PAGERDUTY: "account",
-    Action.Type.OPSGENIE: "account",
-    Action.Type.GITHUB: "integration",
-    Action.Type.GITHUB_ENTERPRISE: "integration",
-    Action.Type.JIRA: "integration",
-    Action.Type.JIRA_SERVER: "integration",
-    Action.Type.AZURE_DEVOPS: "integration",
-}
-
-INTEGRATION_ID_KEY_TO_ACTION_TYPE: dict[str, Action.Type] = {
-    v: k for k, v in ACTION_TYPE_TO_INTEGRATION_ID_KEY.items()
-}
-
-
-ACTION_TYPE_TO_TARGET_IDENTIFIER_KEY: dict[Action.Type, str] = {
-    Action.Type.SLACK: "channel_id",
-    Action.Type.DISCORD: "channel_id",
-    Action.Type.MSTEAMS: "channel_id",
-    Action.Type.PAGERDUTY: "service",
-    Action.Type.OPSGENIE: "team",
-}
-
-TARGET_IDENTIFIER_KEY_TO_ACTION_TYPE: dict[str, Action.Type] = {
-    v: k for k, v in ACTION_TYPE_TO_TARGET_IDENTIFIER_KEY.items()
-}
-
-ACTION_TYPE_TO_TARGET_DISPLAY_KEY: dict[Action.Type, str] = {
-    Action.Type.SLACK: "channel",
-    Action.Type.MSTEAMS: "channel",
-}
-
-TARGET_DISPLAY_KEY_TO_ACTION_TYPE: dict[str, Action.Type] = {
-    v: k for k, v in ACTION_TYPE_TO_TARGET_DISPLAY_KEY.items()
-}
 
+class BaseActionTranslator(ABC):
+    action_type: ClassVar[Action.Type]
+    registry_id: ClassVar[str]
 
+    @property
+    @abstractmethod
+    def target_type(self) -> ActionTarget:
+        pass
+
+    @property
+    @abstractmethod
+    def integration_id_key(self) -> str | None:
+        pass
+
+    @property
+    @abstractmethod
+    def target_identifier_key(self) -> str | None:
+        pass
+
+    @property
+    @abstractmethod
+    def target_display_key(self) -> str | None:
+        pass
+
+    @property
+    @abstractmethod
+    def blob_type(self) -> type["DataBlob"] | None:
+        pass
+
+    def sanitize_action(self, data: dict[str, Any]) -> dict[str, Any]:
+        return data
+
+
+issue_alert_action_translator_registry = Registry[type[BaseActionTranslator]]()
+
+
+@issue_alert_action_translator_registry.register(
+    "sentry.integrations.slack.notify_action.SlackNotifyServiceAction"
+)
+class SlackActionTranslator(BaseActionTranslator):
+    action_type = Action.Type.SLACK
+    registry_id = "sentry.integrations.slack.notify_action.SlackNotifyServiceAction"
+
+    @property
+    def target_type(self) -> ActionTarget:
+        return ActionTarget.SPECIFIC
+
+    @property
+    def integration_id_key(self) -> str:
+        return "workspace"
+
+    @property
+    def target_identifier_key(self) -> str:
+        return "channel_id"
+
+    @property
+    def target_display_key(self) -> str:
+        return "channel"
+
+    @property
+    def blob_type(self) -> type["DataBlob"]:
+        return SlackDataBlob
+
+
+# Keep existing DataBlob classes
 @dataclass
 class DataBlob:
-    """
-    DataBlob is a generic type that represents the data blob for a notification action.
-    """
+    """DataBlob is a generic type that represents the data blob for a notification action."""
 
     pass
 
 
 @dataclass
 class SlackDataBlob(DataBlob):
-    """
-    SlackDataBlob is a specific type that represents the data blob for a Slack notification action.
-    """
+    """SlackDataBlob is a specific type that represents the data blob for a Slack notification action."""
 
     tags: str = ""
     notes: str = ""
-
-
-ACTION_TYPE_2_BLOB_TYPE: dict[Action.Type, type[DataBlob]] = {
-    Action.Type.SLACK: SlackDataBlob,
-}