feat(notif-platform): Prototype the service layer (#93863)

Prototyping the service layer, with some added boilerplate for targets.

They'll need to have the integration data attached by the time we call
`.notify()` so I've added a step in `prepare_targets`, which sidesteps
the frozen data class to ensure they have all the data to send out to
providers.

Also changed `NotificationType` to `NotificationCategory` since type is
a keyword (and a bit overloaded)

Author: leeandher

src/sentry/notifications/platform/discord/provider.py

@@ -7,6 +7,7 @@
 from sentry.notifications.platform.types import (
     NotificationData,
     NotificationProviderKey,
+    NotificationTarget,
     NotificationTargetResourceType,
     NotificationTemplate,
 )
@@ -20,7 +21,9 @@ class DiscordRenderer(NotificationRenderer[DiscordRenderable]):
     provider_key = NotificationProviderKey.DISCORD
 
     @classmethod
-    def render(cls, *, data: NotificationData, template: NotificationTemplate) -> DiscordRenderable:
+    def render[
+        DataT: NotificationData
+    ](cls, *, data: DataT, template: NotificationTemplate[DataT]) -> DiscordRenderable:
         return {}
 
 
@@ -38,3 +41,7 @@ class DiscordNotificationProvider(NotificationProvider[DiscordRenderable]):
     def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
         # TODO(ecosystem): Check for the integration, maybe a feature as well
         return False
+
+    @classmethod
+    def send(cls, *, target: NotificationTarget, renderable: DiscordRenderable) -> None:
+        pass

src/sentry/notifications/platform/email/provider.py

@@ -3,10 +3,11 @@
 from sentry.notifications.platform.provider import NotificationProvider
 from sentry.notifications.platform.registry import provider_registry
 from sentry.notifications.platform.renderer import NotificationRenderer
-from sentry.notifications.platform.target import NotificationTarget
+from sentry.notifications.platform.target import GenericNotificationTarget
 from sentry.notifications.platform.types import (
     NotificationData,
     NotificationProviderKey,
+    NotificationTarget,
     NotificationTargetResourceType,
     NotificationTemplate,
 )
@@ -20,17 +21,23 @@ class EmailRenderer(NotificationRenderer[EmailRenderable]):
     provider_key = NotificationProviderKey.EMAIL
 
     @classmethod
-    def render(cls, *, data: NotificationData, template: NotificationTemplate) -> EmailRenderable:
+    def render[
+        DataT: NotificationData
+    ](cls, *, data: DataT, template: NotificationTemplate[DataT]) -> EmailRenderable:
         return {}
 
 
 @provider_registry.register(NotificationProviderKey.EMAIL)
 class EmailNotificationProvider(NotificationProvider[EmailRenderable]):
     key = NotificationProviderKey.EMAIL
     default_renderer = EmailRenderer
-    target_class = NotificationTarget
+    target_class = GenericNotificationTarget
     target_resource_types = [NotificationTargetResourceType.EMAIL]
 
     @classmethod
     def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
         return True
+
+    @classmethod
+    def send(cls, *, target: NotificationTarget, renderable: EmailRenderable) -> None:
+        pass

src/sentry/notifications/platform/msteams/provider.py

@@ -7,6 +7,7 @@
 from sentry.notifications.platform.types import (
     NotificationData,
     NotificationProviderKey,
+    NotificationTarget,
     NotificationTargetResourceType,
     NotificationTemplate,
 )
@@ -20,7 +21,9 @@ class MSTeamsRenderer(NotificationRenderer[MSTeamsRenderable]):
     provider_key = NotificationProviderKey.MSTEAMS
 
     @classmethod
-    def render(cls, *, data: NotificationData, template: NotificationTemplate) -> MSTeamsRenderable:
+    def render[
+        DataT: NotificationData
+    ](cls, *, data: DataT, template: NotificationTemplate[DataT]) -> MSTeamsRenderable:
         return {}
 
 
@@ -38,3 +41,7 @@ class MSTeamsNotificationProvider(NotificationProvider[MSTeamsRenderable]):
     def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
         # TODO(ecosystem): Check for the integration, maybe a feature as well
         return False
+
+    @classmethod
+    def send(cls, *, target: NotificationTarget, renderable: MSTeamsRenderable) -> None:
+        pass

src/sentry/notifications/platform/provider.py

@@ -1,16 +1,14 @@
-from typing import TYPE_CHECKING, Protocol
+from typing import Protocol
 
+from sentry.notifications.platform.renderer import NotificationRenderer
 from sentry.notifications.platform.types import (
+    NotificationCategory,
     NotificationProviderKey,
+    NotificationTarget,
     NotificationTargetResourceType,
-    NotificationType,
 )
 from sentry.organizations.services.organization.model import RpcOrganizationSummary
 
-if TYPE_CHECKING:
-    from sentry.notifications.platform.renderer import NotificationRenderer
-    from sentry.notifications.platform.target import NotificationTarget
-
 
 class NotificationProviderError(Exception):
     pass
@@ -25,12 +23,12 @@ class NotificationProvider[RenderableT](Protocol):
     """
 
     key: NotificationProviderKey
-    default_renderer: type["NotificationRenderer[RenderableT]"]
-    target_class: type["NotificationTarget"]
+    default_renderer: type[NotificationRenderer[RenderableT]]
+    target_class: type[NotificationTarget]
     target_resource_types: list[NotificationTargetResourceType]
 
     @classmethod
-    def validate_target(cls, *, target: "NotificationTarget") -> None:
+    def validate_target(cls, *, target: NotificationTarget) -> None:
         """
         Validates that a given target is permissible for the provider.
         """
@@ -49,13 +47,14 @@ def validate_target(cls, *, target: "NotificationTarget") -> None:
                 f"Target with resource type '{target.resource_type}' is not supported by {cls.__name__}"
                 f"Supported resource types: {', '.join(t.value for t in cls.target_resource_types)}"
             )
-
         return
 
     @classmethod
-    def get_renderer(cls, *, type: NotificationType) -> type["NotificationRenderer[RenderableT]"]:
+    def get_renderer(
+        cls, *, category: NotificationCategory
+    ) -> type[NotificationRenderer[RenderableT]]:
         """
-        Returns the renderer for a given notification type, falling back to the default renderer.
+        Returns an instance of a renderer for a given notification, falling back to the default renderer.
         Override this to method to permit different renderers for the provider, though keep in mind
         that this may produce inconsistencies between notifications.
         """
@@ -67,3 +66,10 @@ def is_available(cls, *, organization: RpcOrganizationSummary | None = None) ->
         Returns `True` if the provider is available given the key word arguments.
         """
         ...
+
+    @classmethod
+    def send(cls, *, target: NotificationTarget, renderable: RenderableT) -> None:
+        """
+        Using the renderable format for the provider, send a notification to the target.
+        """
+        ...

src/sentry/notifications/platform/renderer.py

@@ -20,7 +20,9 @@ class NotificationRenderer[RenderableT](Protocol):
     provider_key: NotificationProviderKey
 
     @classmethod
-    def render(cls, *, data: NotificationData, template: NotificationTemplate) -> RenderableT:
+    def render[
+        DataT: NotificationData
+    ](cls, *, data: DataT, template: NotificationTemplate[DataT]) -> RenderableT:
         """
         Convert template, and data into a renderable object.
         The form of the renderable object is defined by the provider.

src/sentry/notifications/platform/service.py

@@ -0,0 +1,83 @@
+import logging
+from typing import Final
+
+from sentry.notifications.platform.registry import provider_registry
+from sentry.notifications.platform.target import prepare_targets
+from sentry.notifications.platform.types import (
+    NotificationData,
+    NotificationStrategy,
+    NotificationTarget,
+    NotificationTemplate,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class NotificationServiceError(Exception):
+    pass
+
+
+class NotificationService[T: NotificationData]:
+    def __init__(self, *, data: T):
+        self.data: Final[T] = data
+
+    def notify_prepared_target(
+        self,
+        *,
+        target: NotificationTarget,
+        template: NotificationTemplate[T],
+    ) -> None:
+        """
+        Send a notification directly to a prepared target.
+        NOTE: This method ignores notification settings. When possible, consider using a strategy instead of
+        using this method directly to prevent unwanted noise associated with your notifications.
+        """
+        if not self.data:
+            raise NotificationServiceError(
+                "Notification service must be initialized with data before sending!"
+            )
+
+        # Step 1: Ensure the target has already been prepared.
+        if not target.is_prepared:
+            raise NotificationServiceError(
+                "Target must have `prepare_targets` called prior to sending"
+            )
+
+        # Step 3: Get the provider, and validate the target against it
+        provider = provider_registry.get(target.provider_key)
+        provider.validate_target(target=target)
+
+        # Step 4: Render the template
+        renderer = provider.get_renderer(category=self.data.category)
+        renderable = renderer.render(data=self.data, template=template)
+
+        # Step 5: Send the notification
+        provider.send(target=target, renderable=renderable)
+
+    def notify(
+        self,
+        *,
+        strategy: NotificationStrategy | None = None,
+        targets: list[NotificationTarget] | None = None,
+        template: NotificationTemplate[T],
+    ) -> None:
+
+        if not strategy and not targets:
+            raise NotificationServiceError(
+                "Must provide either a strategy or targets. Strategy is preferred."
+            )
+        if strategy and targets:
+            raise NotificationServiceError(
+                "Cannot provide both strategy and targets, only one is permitted. Strategy is preferred."
+            )
+        if strategy:
+            targets = strategy.get_targets()
+        if not targets:
+            logger.info("Strategy '%s' did not yield targets", strategy.__class__.__name__)
+            return
+
+        # Prepare the targets for sending by fetching integration data, etc.
+        prepare_targets(targets=targets)
+
+        for target in targets:
+            self.notify_prepared_target(target=target, template=template)

src/sentry/notifications/platform/slack/provider.py

@@ -7,6 +7,7 @@
 from sentry.notifications.platform.types import (
     NotificationData,
     NotificationProviderKey,
+    NotificationTarget,
     NotificationTargetResourceType,
     NotificationTemplate,
 )
@@ -17,10 +18,12 @@
 
 
 class SlackRenderer(NotificationRenderer[SlackRenderable]):
-    provider_key = NotificationProviderKey.SLACK
+    provider_key = NotificationProviderKey.DISCORD
 
     @classmethod
-    def render(cls, *, data: NotificationData, template: NotificationTemplate) -> SlackRenderable:
+    def render[
+        DataT: NotificationData
+    ](cls, *, data: DataT, template: NotificationTemplate[DataT]) -> SlackRenderable:
         return {}
 
 
@@ -38,3 +41,7 @@ class SlackNotificationProvider(NotificationProvider[SlackRenderable]):
     def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
         # TODO(ecosystem): Check for the integration, maybe a feature as well
         return False
+
+    @classmethod
+    def send(cls, *, target: NotificationTarget, renderable: SlackRenderable) -> None:
+        pass

src/sentry/notifications/platform/target.py

@@ -1,19 +1,29 @@
 from dataclasses import dataclass, field
 from typing import Any
 
+from sentry.integrations.services.integration.model import (
+    RpcIntegration,
+    RpcOrganizationIntegration,
+)
 from sentry.notifications.platform.types import (
     NotificationProviderKey,
+    NotificationTarget,
     NotificationTargetResourceType,
 )
 
 
+class NotificationTargetError(Exception):
+    pass
+
+
 @dataclass(kw_only=True, frozen=True)
-class NotificationTarget:
+class GenericNotificationTarget(NotificationTarget):
     """
     A designated recipient for a notification. This could be a user, a team, or a channel.
     Accepts the renderable object type that matches the connected provider.
     """
 
+    is_prepared: bool = field(init=False, default=False)
     provider_key: NotificationProviderKey
     resource_type: NotificationTargetResourceType
     resource_id: str
@@ -31,11 +41,29 @@ class NotificationTarget:
 
 
 @dataclass(kw_only=True, frozen=True)
-class IntegrationNotificationTarget(NotificationTarget):
+class IntegrationNotificationTarget(GenericNotificationTarget):
     """
     Adds necessary properties and methods to designate a target within an integration.
     Accepts the renderable object type that matches the connected provider.
     """
 
     integration_id: int
     organization_id: int
+    integration: RpcIntegration = field(init=False)
+    """
+    The integration associated with the target, must call `prepare_targets` to populate.
+    """
+    organization_integration: RpcOrganizationIntegration = field(init=False)
+    """
+    The integration associated with the target, must call `prepare_targets` to populate.
+    """
+
+
+def prepare_targets(targets: list[NotificationTarget]) -> None:
+    """
+    This method is used to prepare the targets for receiving a notification.
+    For example, with IntegrationNotificationTargets, this will populate the `integration` and
+    `organization_integration` fields by making RPC/DB calls.
+    """
+    for target in targets:
+        object.__setattr__(target, "is_prepared", True)