Merge branch 'master' into ele-4029-slack-message-blocks

Author: ofek1weiss

elementary/config/config.py

@@ -10,6 +10,8 @@
 from elementary.monitor.alerts.grouping_type import GroupingType
 from elementary.utils.ordered_yaml import OrderedYaml
 
+DEFAULT_ENV = "dev"
+
 
 class Config:
     _SLACK = "slack"
@@ -68,7 +70,7 @@ def __init__(
         azure_container_name: Optional[str] = None,
         report_url: Optional[str] = None,
         teams_webhook: Optional[str] = None,
-        env: str = "dev",
+        env: str = DEFAULT_ENV,
         run_dbt_deps_if_needed: Optional[bool] = None,
         project_name: Optional[str] = None,
     ):
@@ -249,6 +251,10 @@ def has_gcloud(self):
     def has_gcs(self):
         return self.gcs_bucket_name and self.has_gcloud
 
+    @property
+    def specified_env(self) -> Optional[str]:
+        return self.env if self.env != DEFAULT_ENV else None
+
     def validate_monitor(self):
         provided_integrations = list(
             filter(

elementary/messages/block_builders.py

@@ -87,6 +87,24 @@ def SummaryLineBlock(
     return LineBlock(inlines=text_blocks)
 
 
+def NonPrimaryFactBlock(fact: Tuple[LineBlock, LineBlock]) -> FactBlock:
+    title, value = fact
+    return FactBlock(
+        title=title,
+        value=value,
+        primary=False,
+    )
+
+
+def PrimaryFactBlock(fact: Tuple[LineBlock, LineBlock]) -> FactBlock:
+    title, value = fact
+    return FactBlock(
+        title=title,
+        value=value,
+        primary=True,
+    )
+
+
 def FactsBlock(
     *,
     facts: Sequence[

elementary/messages/messaging_integrations/README.md

@@ -0,0 +1,58 @@
+# Elementary Messaging Integration System
+
+## Overview
+
+The Elementary Messaging Integration system provides a flexible and extensible framework for sending alerts and messages to various messaging platforms (e.g., Slack, Teams). The system is designed to support a gradual migration from the legacy integration system to a more generic messaging-based approach.
+
+## Architecture
+
+### BaseMessagingIntegration
+
+The core of the new messaging system is the `BaseMessagingIntegration` abstract class. This class defines the contract that all messaging integrations must follow:
+
+- `send_message()`: Send a message to a specific destination
+- `supports_reply()`: Check if the integration supports message threading/replies
+- `reply_to_message()`: Reply to an existing message (if supported)
+
+### Key Components
+
+1. **MessageBody**: A platform-agnostic representation of a message
+2. **MessageSendResult**: Contains information about a sent message, including timestamp and platform-specific context
+3. **DestinationType**: Generic type representing the destination for a message (e.g., webhook URL, channel)
+4. **MessageContextType**: Generic type for platform-specific message context
+
+## Migration Strategy
+
+The system currently supports both:
+
+- Legacy `BaseIntegration` implementations (e.g., Slack)
+- New `BaseMessagingIntegration` implementations (e.g., Teams)
+
+This dual support allows for a gradual migration path where:
+
+1. New integrations are implemented using `BaseMessagingIntegration`
+2. Existing integrations can be migrated one at a time
+3. The legacy `BaseIntegration` will eventually be deprecated
+
+## Implementing a New Integration
+
+To add a new messaging platform integration:
+
+1. Create a new class that extends `BaseMessagingIntegration`
+2. Implement the required abstract methods:
+   ```python
+   def send_message(self, destination: DestinationType, body: MessageBody) -> MessageSendResult
+   def supports_reply(self) -> bool
+   def reply_to_message(self, destination, message_context, message_body) -> MessageSendResult  # if supported
+   ```
+3. Update the `Integrations` factory class to support the new integration
+
+## Current Implementations
+
+- **Teams**: Uses the new `BaseMessagingIntegration` system with webhook support
+- **Slack**: Currently uses the legacy `BaseIntegration` system (planned for migration)
+
+## Future Improvements
+
+1. Complete migration of Slack to `BaseMessagingIntegration`
+2. Add support for more messaging platforms

elementary/messages/messaging_integrations/__init__.py

@@ -0,0 +1,49 @@
+from abc import ABC, abstractmethod
+from datetime import datetime
+from typing import Generic, Optional, TypeVar
+
+from pydantic import BaseModel
+
+from elementary.messages.message_body import MessageBody
+from elementary.messages.messaging_integrations.exceptions import (
+    MessageIntegrationReplyNotSupportedError,
+)
+from elementary.utils.log import get_logger
+
+logger = get_logger(__name__)
+
+
+T = TypeVar("T")
+
+
+class MessageSendResult(BaseModel, Generic[T]):
+    timestamp: datetime
+    message_context: Optional[T] = None
+
+
+DestinationType = TypeVar("DestinationType")
+MessageContextType = TypeVar("MessageContextType")
+
+
+class BaseMessagingIntegration(ABC, Generic[DestinationType, MessageContextType]):
+    @abstractmethod
+    def send_message(
+        self,
+        destination: DestinationType,
+        body: MessageBody,
+    ) -> MessageSendResult[MessageContextType]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def supports_reply(self) -> bool:
+        raise NotImplementedError
+
+    def reply_to_message(
+        self,
+        destination: DestinationType,
+        message_context: MessageContextType,
+        body: MessageBody,
+    ) -> MessageSendResult[MessageContextType]:
+        if not self.supports_reply():
+            raise MessageIntegrationReplyNotSupportedError
+        raise NotImplementedError

elementary/messages/messaging_integrations/base_messaging_integration.py

@@ -0,0 +1,6 @@
+class MessagingIntegrationError(Exception):
+    pass
+
+
+class MessageIntegrationReplyNotSupportedError(MessagingIntegrationError):
+    pass

elementary/messages/messaging_integrations/exceptions.py

@@ -0,0 +1,82 @@
+from datetime import datetime
+from typing import Optional
+
+import requests
+from pydantic import BaseModel
+
+from elementary.messages.formats.adaptive_cards import format_adaptive_card
+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 (
+    MessageIntegrationReplyNotSupportedError,
+    MessagingIntegrationError,
+)
+from elementary.utils.log import get_logger
+
+logger = get_logger(__name__)
+
+
+class ChannelWebhook(BaseModel):
+    webhook: str
+    channel: Optional[str] = None
+
+
+def send_adaptive_card(webhook_url: str, card: dict) -> requests.Response:
+    """Sends an Adaptive Card to the specified webhook URL."""
+    payload = {
+        "type": "message",
+        "attachments": [
+            {
+                "contentType": "application/vnd.microsoft.card.adaptive",
+                "contentUrl": None,
+                "content": card,
+            }
+        ],
+    }
+
+    response = requests.post(
+        webhook_url,
+        json=payload,
+        headers={"Content-Type": "application/json"},
+    )
+    response.raise_for_status()
+    if response.status_code == 202:
+        logger.debug("Got 202 response from Teams webhook, assuming success")
+    return response
+
+
+class TeamsWebhookMessagingIntegration(
+    BaseMessagingIntegration[ChannelWebhook, ChannelWebhook]
+):
+    def send_message(
+        self,
+        destination: ChannelWebhook,
+        body: MessageBody,
+    ) -> MessageSendResult[ChannelWebhook]:
+        card = format_adaptive_card(body)
+        try:
+            send_adaptive_card(destination.webhook, card)
+            return MessageSendResult(
+                message_context=destination,
+                timestamp=datetime.utcnow(),
+            )
+        except requests.RequestException as e:
+            raise MessagingIntegrationError(
+                "Failed to send message to Teams webhook"
+            ) from e
+
+    def supports_reply(self) -> bool:
+        return False
+
+    def reply_to_message(
+        self,
+        destination: ChannelWebhook,
+        message_context: ChannelWebhook,
+        body: MessageBody,
+    ) -> MessageSendResult[ChannelWebhook]:
+        raise MessageIntegrationReplyNotSupportedError(
+            "Teams webhook message integration does not support replying to messages"
+        )

elementary/messages/messaging_integrations/teams_webhook.py

@@ -31,6 +31,7 @@ def __init__(
         report_url: Optional[str] = None,
         alert_fields: Optional[List[str]] = None,
         elementary_database_and_schema: Optional[str] = None,
+        env: Optional[str] = None,
         **kwargs,
     ):
         self.id = id
@@ -63,6 +64,7 @@ def __init__(
         self.report_url = report_url
         self.alert_fields = alert_fields
         self.elementary_database_and_schema = elementary_database_and_schema
+        self.env = env
 
     @property
     def unified_meta(self) -> Dict: