Core 645 generic alert types

elementary/monitor/alerts/alert.py

@@ -99,8 +99,12 @@ def data(self) -> Dict:
         raise NotImplementedError
 
     @property
-    def concise_name(self):
-        return "Alert"
+    def concise_name(self) -> str:
+        raise NotImplementedError
+
+    @property
+    def asset_type(self) -> str:
+        raise NotImplementedError
 
     @property
     def summary(self) -> str:

elementary/monitor/alerts/alert_messages/builder.py

@@ -1,5 +1,5 @@
 from datetime import timedelta
-from typing import Any, Dict, List, Literal, Optional, Sequence, Tuple, Union
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 from elementary.messages.block_builders import (
     BoldTextBlock,
@@ -33,7 +33,7 @@
     TextStyle,
 )
 from elementary.messages.message_body import Color, MessageBlock, MessageBody
-from elementary.monitor.alerts.alert import OrchestratorInfo
+from elementary.monitor.alerts.alert import AlertModel, OrchestratorInfo
 from elementary.monitor.alerts.alert_messages.alert_fields import AlertField
 from elementary.monitor.alerts.alerts_groups.alerts_group import AlertsGroup
 from elementary.monitor.alerts.alerts_groups.base_alerts_group import BaseAlertsGroup
@@ -51,12 +51,7 @@
 )
 from elementary.utils.pydantic_shim import BaseModel
 
-AlertType = Union[
-    TestAlertModel,
-    ModelAlertModel,
-    SourceFreshnessAlertModel,
-    BaseAlertsGroup,
-]
+AlertType = Union[AlertModel, BaseAlertsGroup]
 
 
 class MessageBuilderConfig(BaseModel):
@@ -104,7 +99,7 @@ def _get_alert_title(
 
     def _get_run_alert_subtitle_block(
         self,
-        type: Literal["test", "snapshot", "model", "source"],
+        type: str,
         name: str,
         status: Optional[str] = None,
         detected_at_str: Optional[str] = None,
@@ -172,7 +167,7 @@ def _get_run_alert_subtitle_block(
 
     def _get_run_alert_subtitle_links(
         self,
-        alert: Union[TestAlertModel, SourceFreshnessAlertModel, ModelAlertModel],
+        alert: AlertModel,
     ) -> List[ReportLinkData]:
         report_link = alert.get_report_link()
         if report_link:
@@ -181,25 +176,14 @@ def _get_run_alert_subtitle_links(
 
     def _get_run_alert_subtitle_blocks(
         self,
-        alert: Union[TestAlertModel, SourceFreshnessAlertModel, ModelAlertModel],
+        alert: AlertModel,
     ) -> List[MessageBlock]:
-        asset_type: Literal["test", "snapshot", "model", "source"]
-        asset_name: str
-        if isinstance(alert, TestAlertModel):
-            asset_type = "test"
-            asset_name = alert.concise_name
-        elif isinstance(alert, SourceFreshnessAlertModel):
-            asset_type = "source"
-            asset_name = f"{alert.source_name}.{alert.identifier}"
-        elif isinstance(alert, ModelAlertModel):
-            asset_type = "snapshot" if alert.materialization == "snapshot" else "model"
-            asset_name = alert.alias
         links = self._get_run_alert_subtitle_links(alert)
         orchestrator_info = alert.orchestrator_info
         return [
             self._get_run_alert_subtitle_block(
-                type=asset_type,
-                name=asset_name,
+                type=alert.asset_type,
+                name=alert.concise_name,
                 status=alert.status,
                 detected_at_str=alert.detected_at_str,
                 suppression_interval=alert.suppression_interval,
@@ -472,11 +456,7 @@ def _get_source_freshness_alert_config_blocks(
 
     def _get_alert_list_line(
         self,
-        alert: Union[
-            TestAlertModel,
-            ModelAlertModel,
-            SourceFreshnessAlertModel,
-        ],
+        alert: AlertModel,
     ) -> LineBlock:
         inlines: List[InlineBlock] = [
             TextBlock(text=alert.summary, style=TextStyle.BOLD),
@@ -507,13 +487,7 @@ def _get_alert_list_blocks(
         self,
         title: str,
         bullet_icon: Icon,
-        alerts: Sequence[
-            Union[
-                TestAlertModel,
-                ModelAlertModel,
-                SourceFreshnessAlertModel,
-            ]
-        ],
+        alerts: Sequence[AlertModel],
     ) -> List[MessageBlock]:
         blocks: List[MessageBlock] = []
         if not alerts:
@@ -526,10 +500,10 @@ def _get_alert_list_blocks(
 
     def _get_sub_alert_groups_blocks(
         self,
-        test_errors: List[Union[TestAlertModel, SourceFreshnessAlertModel]],
-        test_warnings: List[Union[TestAlertModel, SourceFreshnessAlertModel]],
-        test_failures: List[Union[TestAlertModel, SourceFreshnessAlertModel]],
-        model_errors: List[ModelAlertModel],
+        test_errors: Sequence[AlertModel],
+        test_warnings: Sequence[AlertModel],
+        test_failures: Sequence[AlertModel],
+        model_errors: Sequence[AlertModel],
     ) -> List[MessageBlock]:
         blocks: List[MessageBlock] = []
         model_errors_alert_list_blocks = self._get_alert_list_blocks(

elementary/monitor/alerts/alerts_groups/alerts_group.py

@@ -1,20 +1,19 @@
-from typing import List, Optional, Union
+from typing import List, Optional, Sequence
 
+from elementary.monitor.alerts.alert import AlertModel
 from elementary.monitor.alerts.alerts_groups.base_alerts_group import BaseAlertsGroup
 from elementary.monitor.alerts.model_alert import ModelAlertModel
-from elementary.monitor.alerts.source_freshness_alert import SourceFreshnessAlertModel
-from elementary.monitor.alerts.test_alert import TestAlertModel
 
 
 class AlertsGroup(BaseAlertsGroup):
-    test_errors: List[Union[TestAlertModel, SourceFreshnessAlertModel]]
-    test_warnings: List[Union[TestAlertModel, SourceFreshnessAlertModel]]
-    test_failures: List[Union[TestAlertModel, SourceFreshnessAlertModel]]
+    test_errors: List[AlertModel]
+    test_warnings: List[AlertModel]
+    test_failures: List[AlertModel]
     model_errors: List[ModelAlertModel]
 
     def __init__(
         self,
-        alerts: List[Union[TestAlertModel, ModelAlertModel, SourceFreshnessAlertModel]],
+        alerts: Sequence[AlertModel],
         env: Optional[str] = None,
     ) -> None:
         super().__init__(alerts, env=env)

elementary/monitor/alerts/alerts_groups/base_alerts_group.py

@@ -1,18 +1,14 @@
 from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import Dict, List, Optional, Sequence, Union
+from typing import Dict, List, Optional, Sequence
 
-from elementary.monitor.alerts.model_alert import ModelAlertModel
-from elementary.monitor.alerts.source_freshness_alert import SourceFreshnessAlertModel
-from elementary.monitor.alerts.test_alert import TestAlertModel
+from elementary.monitor.alerts.alert import AlertModel
 
 
 class BaseAlertsGroup(ABC):
     def __init__(
         self,
-        alerts: Sequence[
-            Union[TestAlertModel, ModelAlertModel, SourceFreshnessAlertModel]
-        ],
+        alerts: Sequence[AlertModel],
         env: Optional[str] = None,
     ) -> None:
         self.alerts = alerts

elementary/monitor/alerts/model_alert.py

@@ -100,13 +100,13 @@ def data(self) -> Dict:
             full_refresh=self.full_refresh,
         )
 
+    @property
+    def asset_type(self) -> str:
+        return "snapshot" if self.materialization == "snapshot" else "model"
+
     @property
     def concise_name(self) -> str:
-        if self.materialization == "snapshot":
-            dbt_type = "snapshot"
-        else:
-            dbt_type = "model"
-        return f"dbt {dbt_type} alert - {self.alias}"
+        return self.alias
 
     @property
     def summary(self) -> str:

elementary/monitor/alerts/source_freshness_alert.py

@@ -155,9 +155,13 @@ def data(self) -> Dict:
             freshness_description=self.freshness_description,
         )
 
+    @property
+    def asset_type(self) -> str:
+        return "source"
+
     @property
     def concise_name(self) -> str:
-        return f"source freshness alert - {self.source_name}.{self.identifier}"
+        return f"{self.source_name}.{self.identifier}"
 
     @property
     def error_message(self) -> str:

elementary/monitor/alerts/test_alert.py

@@ -166,6 +166,10 @@ def data(self) -> Dict:
             env=self.env,
         )
 
+    @property
+    def asset_type(self) -> str:
+        return "test"
+
     @property
     def concise_name(self) -> str:
         if self.test_sub_type_display_name.lower() not in (

elementary/monitor/api/alerts/alert_filters.py

@@ -29,7 +29,7 @@ def _get_alert_node_name(alert: PendingAlertSchema) -> Optional[str]:
     alert_node_name = None
     alert_type = AlertTypes(alert.type)
     if alert_type is AlertTypes.TEST:
-        alert_node_name = alert.data.test_name  # type: ignore[union-attr]
+        alert_node_name = alert.data.test_name  # type: ignore[attr-defined]
     elif alert_type is AlertTypes.MODEL or alert_type is AlertTypes.SOURCE_FRESHNESS:
         alert_node_name = alert.data.model_unique_id
     else:

elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py

@@ -22,6 +22,7 @@
 from elementary.messages.messaging_integrations.exceptions import (
     MessagingIntegrationError,
 )
+from elementary.monitor.alerts.alert import AlertModel
 from elementary.monitor.alerts.alert_messages.builder import (
     AlertMessageBuilder,
     MessageBuilderConfig,
@@ -385,13 +386,7 @@ def _send_alerts(
             self.execution_properties["sent_alert_count"] = self.sent_alert_count
             return
 
-        sent_successfully_alerts: List[
-            Union[
-                TestAlertModel,
-                ModelAlertModel,
-                SourceFreshnessAlertModel,
-            ]
-        ] = []
+        sent_successfully_alerts: List[AlertModel] = []
 
         with alive_bar(len(alerts), title="Sending alerts") as bar:
             for alert in alerts:

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

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 from typing import Union
 
+from elementary.monitor.alerts.alert import AlertModel
 from elementary.monitor.alerts.alerts_groups import GroupedByTableAlerts
 from elementary.monitor.alerts.alerts_groups.base_alerts_group import BaseAlertsGroup
 from elementary.monitor.alerts.model_alert import ModelAlertModel
@@ -21,13 +22,7 @@ def _initial_client(self, *args, **kwargs):
 
     def _get_alert_template(
         self,
-        alert: Union[
-            TestAlertModel,
-            ModelAlertModel,
-            SourceFreshnessAlertModel,
-            GroupedByTableAlerts,
-            BaseAlertsGroup,
-        ],
+        alert: Union[AlertModel, GroupedByTableAlerts, BaseAlertsGroup],
         *args,
         **kwargs,
     ):
@@ -83,12 +78,7 @@ def _get_alerts_group_template(self, alert: BaseAlertsGroup, *args, **kwargs):
     @abstractmethod
     def _get_fallback_template(
         self,
-        alert: Union[
-            TestAlertModel,
-            ModelAlertModel,
-            SourceFreshnessAlertModel,
-            GroupedByTableAlerts,
-        ],
+        alert: Union[AlertModel, GroupedByTableAlerts],
         *args,
         **kwargs,
     ):
@@ -97,13 +87,7 @@ def _get_fallback_template(
     @abstractmethod
     def send_alert(
         self,
-        alert: Union[
-            TestAlertModel,
-            ModelAlertModel,
-            SourceFreshnessAlertModel,
-            GroupedByTableAlerts,
-            BaseAlertsGroup,
-        ],
+        alert: Union[AlertModel, GroupedByTableAlerts, BaseAlertsGroup],
         *args,
         **kwargs,
     ) -> bool: