Add support for new messaging integration in data monitoring alerts

- Updated `DataMonitoringAlerts` to support both legacy and new messaging integrations
- Implemented dual-path alert sending logic for BaseIntegration and BaseMessagingIntegration
- Added `_send_message()` method to handle new messaging integration message sending
- Created `get_health_check_message()` for test message support in new integrations
- Updated `Integrations` class to support gradual migration to new messaging system
- Added destination resolution method for messaging integrations

Author: MikaKerman

elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py

@@ -6,9 +6,19 @@
 from alive_progress import alive_bar
 
 from elementary.config.config import Config
+from elementary.messages.block_builders import TextLineBlock
+from elementary.messages.blocks import HeaderBlock, LinesBlock
+from elementary.messages.message_body import MessageBody
+from elementary.messages.messaging_integrations.base_messaging_integration import (
+    BaseMessagingIntegration,
+    MessageSendResult,
+)
+from elementary.messages.messaging_integrations.exceptions import (
+    MessagingIntegrationError,
+)
+from elementary.monitor.alerts.alert_messages.builder import AlertMessageBuilder
 from elementary.monitor.alerts.alerts_groups import GroupedByTableAlerts
 from elementary.monitor.alerts.alerts_groups.alerts_group import AlertsGroup
-from elementary.monitor.alerts.alerts_groups.base_alerts_group import BaseAlertsGroup
 from elementary.monitor.alerts.grouping_type import GroupingType
 from elementary.monitor.alerts.model_alert import ModelAlertModel
 from elementary.monitor.alerts.source_freshness_alert import SourceFreshnessAlertModel
@@ -32,7 +42,26 @@
 logger = get_logger(__name__)
 
 
+def get_health_check_message() -> MessageBody:
+    return MessageBody(
+        blocks=[
+            HeaderBlock(text="Elementary monitor ran successfully"),
+            LinesBlock(
+                lines=[
+                    TextLineBlock(
+                        text=f"Elementary monitor ran successfully on {datetime.now().strftime('%Y-%m-%d %H:%M')}"
+                    ),
+                ]
+            ),
+        ]
+    )
+
+
 class DataMonitoringAlerts(DataMonitoring):
+    # The alerts_integration field now supports both the legacy BaseIntegration and the new BaseMessagingIntegration
+    # This dual support allows for a gradual migration from the old integration system to the new messaging system
+    alerts_integration: Union[BaseIntegration, BaseMessagingIntegration]
+
     def __init__(
         self,
         config: Config,
@@ -61,7 +90,9 @@ def __init__(
         self.override_config_defaults = override_config
         self.alerts_integration = self._get_integration_client()
 
-    def _get_integration_client(self) -> BaseIntegration:
+    def _get_integration_client(
+        self,
+    ) -> Union[BaseIntegration, BaseMessagingIntegration]:
         return Integrations.get_integration(
             config=self.config,
             tracking=self.tracking,
@@ -183,7 +214,7 @@ def _format_alerts(
             ModelAlertModel,
             SourceFreshnessAlertModel,
             GroupedByTableAlerts,
-            BaseAlertsGroup,
+            AlertsGroup,
         ]
     ]:
         group_all_alerts = len(alerts) >= self.config.group_alerts_threshold  # type: ignore[arg-type]
@@ -244,11 +275,57 @@ def _format_alerts(
                 key=lambda alert: alert.detected_at or datetime.max,
             )
 
-    def _send_test_message(self):
-        self.alerts_integration.send_test_message(
-            channel_name=self.config.slack_channel_name
+    def _send_message(
+        self, integration: BaseMessagingIntegration, message_body: MessageBody
+    ) -> MessageSendResult:
+        destination = Integrations.get_destination(
+            integration=integration, config=self.config
+        )
+        return integration.send_message(
+            destination=destination, message_body=message_body
         )
 
+    def _send_test_message(self):
+        if isinstance(self.alerts_integration, BaseIntegration):
+            self.alerts_integration.send_test_message(
+                channel_name=self.config.slack_channel_name
+            )
+        else:
+            test_message = get_health_check_message()
+            return self._send_message(
+                integration=self.alerts_integration, message_body=test_message
+            )
+
+    def _send_alert(
+        self,
+        alert: Union[
+            TestAlertModel,
+            ModelAlertModel,
+            SourceFreshnessAlertModel,
+            GroupedByTableAlerts,
+            AlertsGroup,
+        ],
+    ):
+        # Support both legacy BaseIntegration and new BaseMessagingIntegration
+        # BaseIntegration will be deprecated in favor of BaseMessagingIntegration
+        if isinstance(self.alerts_integration, BaseIntegration):
+            return self.alerts_integration.send_alert(alert)
+        else:
+            # New messaging integration path - converts alerts to message bodies
+            alert_message_builder = AlertMessageBuilder()
+            alert_message_body = alert_message_builder.build(
+                alert=alert,
+            )
+            try:
+                self._send_message(
+                    integration=self.alerts_integration,
+                    message_body=alert_message_body,
+                )
+                return True
+            except MessagingIntegrationError:
+                logger.error(f"Could not send the alert - {type(alert)}.")
+                return False
+
     def _send_alerts(
         self,
         alerts: List[
@@ -257,7 +334,7 @@ def _send_alerts(
                 ModelAlertModel,
                 SourceFreshnessAlertModel,
                 GroupedByTableAlerts,
-                BaseAlertsGroup,
+                AlertsGroup,
             ]
         ],
     ):
@@ -275,15 +352,15 @@ def _send_alerts(
 
         with alive_bar(len(alerts), title="Sending alerts") as bar:
             for alert in alerts:
-                sent_successfully = self.alerts_integration.send_alert(alert)
+                sent_successfully = self._send_alert(alert)
                 bar()
                 if sent_successfully:
-                    if isinstance(alert, BaseAlertsGroup):
+                    if isinstance(alert, AlertsGroup):
                         sent_successfully_alerts.extend(alert.alerts)
                     else:
                         sent_successfully_alerts.append(alert)
                 else:
-                    if isinstance(alert, BaseAlertsGroup):
+                    if isinstance(alert, AlertsGroup):
                         for inner_alert in alert.alerts:
                             logger.error(
                                 f"Could not send the alert - {inner_alert.id}. Full alert: {json.dumps(inner_alert.data)}"

elementary/monitor/data_monitoring/alerts/integrations/integrations.py

@@ -1,16 +1,20 @@
-from typing import Optional
+from typing import Any, Optional, Union
 
 from elementary.config.config import Config
 from elementary.exceptions.exceptions import Error
+from elementary.messages.messaging_integrations.base_messaging_integration import (
+    BaseMessagingIntegration,
+)
+from elementary.messages.messaging_integrations.teams_webhook import (
+    ChannelWebhook,
+    TeamsWebhookMessagingIntegration,
+)
 from elementary.monitor.data_monitoring.alerts.integrations.base_integration import (
     BaseIntegration,
 )
 from elementary.monitor.data_monitoring.alerts.integrations.slack.slack import (
     SlackIntegration,
 )
-from elementary.monitor.data_monitoring.alerts.integrations.teams.teams import (
-    TeamsIntegration,
-)
 from elementary.tracking.tracking_interface import Tracking
 
 
@@ -29,18 +33,31 @@ def get_integration(
         config: Config,
         tracking: Optional[Tracking] = None,
         override_config_defaults: bool = False,
-    ) -> BaseIntegration:
+    ) -> Union[BaseIntegration, BaseMessagingIntegration]:
+        # Factory method that returns either a legacy BaseIntegration or new BaseMessagingIntegration
+        # This allows for a gradual migration from the old integration system to the new messaging system
+        # - Slack currently uses the legacy BaseIntegration
+        # - Teams uses the new BaseMessagingIntegration
         if config.has_slack:
             return SlackIntegration(
                 config=config,
                 tracking=tracking,
                 override_config_defaults=override_config_defaults,
             )
         elif config.has_teams:
-            return TeamsIntegration(
-                config=config,
-                tracking=tracking,
-                override_config_defaults=override_config_defaults,
-            )
+            return TeamsWebhookMessagingIntegration()
         else:
             raise UnsupportedAlertIntegrationError
+
+    @staticmethod
+    def get_destination(integration: BaseMessagingIntegration, config: Config) -> Any:
+        # Helper method to get the appropriate destination for BaseMessagingIntegration implementations
+        # Each messaging integration type may have different destination requirements
+        # Currently supports Teams webhook destinations
+        if (
+            isinstance(integration, TeamsWebhookMessagingIntegration)
+            and config.has_teams
+            and config.teams_webhook
+        ):
+            return ChannelWebhook(webhook=config.teams_webhook)
+        return None