feat(notif-platform): Implement a notification provider registry (#93119)

Follows the scheme from workflow of using a decorator for registration.
Registers the classes behind the providers rather than instances (as
described in the spec).

I did see some existing types for a few registrations (Slack and
MSTeams) but since this they're loaded on app init, they were having
circular import issues. Gonna work through that when I start digging
into individual providers, so skipping it for now,.

Author: leeandher

pyproject.toml

@@ -335,6 +335,7 @@ module = [
     "sentry.nodestore.django.models",
     "sentry.nodestore.filesystem.backend",
     "sentry.nodestore.models",
+    "sentry.notifications.platform.*",
     "sentry.notifications.services.*",
     "sentry.options.rollout",
     "sentry.organizations.*",

src/sentry/conf/server.py

@@ -408,6 +408,7 @@ def env(
     "sentry.sentry_apps",
     "sentry.integrations",
     "sentry.notifications",
+    "sentry.notifications.platform",
     "sentry.flags",
     "sentry.monitors",
     "sentry.uptime",

src/sentry/notifications/platform/README.md

@@ -0,0 +1,16 @@
+# Notification Platform
+
+TODO(ecosystem): Fill this out
+
+## NotificationProvider
+
+A notification provider is the system which Sentry will trigger in order to notify NotificationTargets, for example; Email or Slack.
+
+To register a new provider:
+
+1. Add a key to `NotificationProviderKey` in [`.types`](./types.py)
+2. Add a directory for your new provider.
+3. Create a provider module in the new directory.
+   - Extend `NotificationProvider` from [`.provider`](./provider.py) and implement its methods/variables.
+   - Import `provider_registry` from [`.registry`](./registry.py) and add it via decorator: `@provider_registry.register(<YOUR-KEY-HERE>)`
+4. In [.apps](./apps.py), explicitly import the module to register the provider on initialization.

src/sentry/notifications/platform/__init__.py

@@ -0,0 +1,12 @@
+from django.apps import AppConfig
+
+
+class Config(AppConfig):
+    name = "sentry.notifications.platform"
+
+    def ready(self) -> None:
+        # Register the providers providers
+        from .discord.provider import DiscordNotificationProvider  # NOQA
+        from .email.provider import EmailNotificationProvider  # NOQA
+        from .msteams.provider import MSTeamsNotificationProvider  # NOQA
+        from .slack.provider import SlackNotificationProvider  # NOQA

src/sentry/notifications/platform/apps.py

@@ -0,0 +1,25 @@
+from typing import Any
+
+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.types import NotificationProviderKey
+from sentry.organizations.services.organization.model import RpcOrganizationSummary
+
+# TODO(ecosystem): Proper typing - https://discord.com/developers/docs/resources/message#create-message
+type DiscordRenderable = Any
+
+
+class DiscordRenderer(NotificationRenderer[DiscordRenderable]):
+    provider_key = NotificationProviderKey.DISCORD
+
+
+@provider_registry.register(NotificationProviderKey.DISCORD)
+class DiscordNotificationProvider(NotificationProvider[DiscordRenderable]):
+    key = NotificationProviderKey.DISCORD
+    default_renderer = DiscordRenderer
+
+    @classmethod
+    def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
+        # TODO(ecosystem): Check for the integration, maybe a feature as well
+        return False

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

@@ -0,0 +1,24 @@
+from typing import Any
+
+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.types import NotificationProviderKey
+from sentry.organizations.services.organization.model import RpcOrganizationSummary
+
+# TODO(ecosystem): Proper typing for email payloads (HTML + txt)
+type EmailRenderable = Any
+
+
+class EmailRenderer(NotificationRenderer[EmailRenderable]):
+    provider_key = NotificationProviderKey.EMAIL
+
+
+@provider_registry.register(NotificationProviderKey.EMAIL)
+class EmailNotificationProvider(NotificationProvider[EmailRenderable]):
+    key = NotificationProviderKey.EMAIL
+    default_renderer = EmailRenderer
+
+    @classmethod
+    def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
+        return True

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

@@ -0,0 +1,25 @@
+from typing import Any
+
+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.types import NotificationProviderKey
+from sentry.organizations.services.organization.model import RpcOrganizationSummary
+
+# TODO(ecosystem): Figure out a way to use 'AdaptiveCard' type
+type MSTeamsRenderable = Any
+
+
+class MSTeamsRenderer(NotificationRenderer[MSTeamsRenderable]):
+    provider_key = NotificationProviderKey.MSTEAMS
+
+
+@provider_registry.register(NotificationProviderKey.MSTEAMS)
+class MSTeamsNotificationProvider(NotificationProvider[MSTeamsRenderable]):
+    key = NotificationProviderKey.MSTEAMS
+    default_renderer = MSTeamsRenderer
+
+    @classmethod
+    def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
+        # TODO(ecosystem): Check for the integration, maybe a feature as well
+        return False

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

@@ -0,0 +1,26 @@
+from typing import TYPE_CHECKING, Protocol
+
+from sentry.notifications.platform.types import NotificationProviderKey
+from sentry.organizations.services.organization.model import RpcOrganizationSummary
+
+if TYPE_CHECKING:
+    from sentry.notifications.platform.renderer import NotificationRenderer
+
+
+class NotificationProvider[T](Protocol):
+    """
+    A protocol metaclass for all notification providers.
+
+    Accepts a renderable object type that is understood by the notification provider.
+    For example, Email might expect HTML, or raw text; Slack might expect a JSON Block Kit object.
+    """
+
+    key: NotificationProviderKey
+    default_renderer: type["NotificationRenderer[T]"]
+
+    @classmethod
+    def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
+        """
+        Returns `True` if the provider is available given the key word arguments.
+        """
+        ...

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

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sentry.notifications.platform.provider import NotificationProvider
+from sentry.organizations.services.organization.model import RpcOrganizationSummary
+from sentry.utils.registry import Registry
+
+
+class NotificationProviderRegistry(Registry[type[NotificationProvider[Any]]]):
+    """
+    A registry for notification providers. Adds `get_all` and `get_available` methods to the base registry.
+    """
+
+    def get_all(self) -> list[type[NotificationProvider[Any]]]:
+        """
+        Returns every NotificationProvider that has been registered. Some providers may not be
+        available generally available to all customers. For only released providers, use `get_available` instead.
+        """
+        return list(self.registrations.values())
+
+    def get_available(
+        self, *, organization: RpcOrganizationSummary | None = None
+    ) -> list[type[NotificationProvider[Any]]]:
+        """
+        Returns every registered NotificationProvider that has been released to all customers.
+        """
+        return [
+            provider
+            for provider in self.registrations.values()
+            if provider.is_available(organization=organization)
+        ]
+
+
+provider_registry = NotificationProviderRegistry()

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

@@ -0,0 +1,27 @@
+from typing import Any, Protocol
+
+from sentry.notifications.platform.types import NotificationProviderKey
+
+type NotificationTemplate = Any
+type NotificationData = Any
+
+
+# TODO(ecosystem): Evaluate whether or not this even makes sense as a protocol, or we can just use a typed Callable.
+# If there is only one method, and the class usage is just to call a method, the Callable route might make more sense.
+# The typing T is also sketchy being in only the return position, and not inherently connected to the provider class.
+# The concept of renderers could just be a subset of functionality on the base provider class.
+class NotificationRenderer[T](Protocol):
+    """
+    A protocol metaclass for all notification renderers.
+    Accepts the renderable object type that matches the connected provider.
+    """
+
+    provider_key: NotificationProviderKey
+
+    @classmethod
+    def render(cls, *, data: NotificationData, template: NotificationTemplate) -> T:
+        """
+        Convert template, and data into a renderable object.
+        The form of the renderable object is defined by the provider.
+        """
+        ...

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

@@ -0,0 +1,25 @@
+from typing import Any
+
+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.types import NotificationProviderKey
+from sentry.organizations.services.organization.model import RpcOrganizationSummary
+
+# TODO(ecosystem): Figure out a way to use 'SlackBlock' type
+type SlackRenderable = Any
+
+
+class SlackRenderer(NotificationRenderer[SlackRenderable]):
+    provider_key = NotificationProviderKey.SLACK
+
+
+@provider_registry.register(NotificationProviderKey.SLACK)
+class SlackNotificationProvider(NotificationProvider[SlackRenderable]):
+    key = NotificationProviderKey.SLACK
+    default_renderer = SlackRenderer
+
+    @classmethod
+    def is_available(cls, *, organization: RpcOrganizationSummary | None = None) -> bool:
+        # TODO(ecosystem): Check for the integration, maybe a feature as well
+        return False

src/sentry/notifications/platform/provider.py

@@ -0,0 +1,14 @@
+from enum import StrEnum
+
+from sentry.integrations.types import ExternalProviderEnum
+
+
+class NotificationProviderKey(StrEnum):
+    """
+    The unique keys for each registered notification provider.
+    """
+
+    EMAIL = ExternalProviderEnum.EMAIL
+    SLACK = ExternalProviderEnum.SLACK
+    MSTEAMS = ExternalProviderEnum.MSTEAMS
+    DISCORD = ExternalProviderEnum.DISCORD

src/sentry/notifications/platform/registry.py

@@ -0,0 +1,22 @@
+from sentry.notifications.platform.registry import provider_registry
+from sentry.notifications.platform.types import NotificationProviderKey
+from sentry.organizations.services.organization.serial import serialize_organization_summary
+from sentry.testutils.cases import TestCase
+
+
+class NotificationProviderTest(TestCase):
+    def test_all_registrants_follow_protocol(self):
+        for provider in provider_registry.get_all():
+            # Ensures the provider can be instantiated, does not test functionality
+            provider()
+            # Ensures protocol properties are present and correct
+            assert provider.key in NotificationProviderKey
+            # Ensures the default renderer links back to its connected provider key
+            assert provider.default_renderer.provider_key == provider.key
+            assert isinstance(provider.is_available(), bool)
+            assert isinstance(
+                provider.is_available(
+                    organization=serialize_organization_summary(self.organization)
+                ),
+                bool,
+            )

src/sentry/notifications/platform/renderer.py

@@ -0,0 +1,29 @@
+from sentry.notifications.platform.discord.provider import DiscordNotificationProvider
+from sentry.notifications.platform.email.provider import EmailNotificationProvider
+from sentry.notifications.platform.msteams.provider import MSTeamsNotificationProvider
+from sentry.notifications.platform.registry import provider_registry
+from sentry.notifications.platform.slack.provider import SlackNotificationProvider
+from sentry.testutils.cases import TestCase
+
+
+class NotificationProviderRegistryTest(TestCase):
+    def test_get_all(self):
+        providers = provider_registry.get_all()
+        expected_providers = [
+            EmailNotificationProvider,
+            SlackNotificationProvider,
+            MSTeamsNotificationProvider,
+            DiscordNotificationProvider,
+        ]
+
+        assert len(providers) == len(expected_providers)
+        for provider in expected_providers:
+            assert provider in providers
+
+    def test_get_available(self):
+        providers = provider_registry.get_available()
+        expected_providers = [EmailNotificationProvider]
+
+        assert len(providers) == len(expected_providers)
+        for provider in expected_providers:
+            assert provider in providers