feat: Group alerts by owner

elementary/cli/cli.py

@@ -38,12 +38,12 @@ class ElementaryCLI(click.MultiCommand):
         "run-operation": run_operation,
     }
 
-    def list_commands(self, ctx):
-        return self._CMD_MAP.keys()
+    def list_commands(self, ctx) -> list[str]:
+        return list(self._CMD_MAP.keys())
 
-    def get_command(self, ctx, name):
+    def get_command(self, ctx, cmd_name):
         ctx.auto_envvar_prefix = "EDR"
-        return self._CMD_MAP.get(name)
+        return self._CMD_MAP.get(cmd_name)
 
     def format_help(self, ctx, formatter):
         try:

elementary/config/config.py

@@ -1,6 +1,6 @@
 import os
 from pathlib import Path
-from typing import Optional
+from typing import Optional, cast
 
 import google.auth  # type: ignore[import]
 from dateutil import tz
@@ -88,11 +88,11 @@ def __init__(
 
         config = self._load_configuration()
 
-        self.target_dir = self._first_not_none(
+        self.target_dir = str(self._first_not_none(
             target_path,
             config.get("target-path"),
             os.getcwd(),
-        )
+        ))
         os.makedirs(os.path.abspath(self.target_dir), exist_ok=True)
         os.environ["DBT_LOG_PATH"] = os.path.abspath(target_path)
 
@@ -129,11 +129,11 @@ def __init__(
             slack_config.get("group_alerts_by"),
             GroupingType.BY_ALERT.value,
         )
-        self.group_alerts_threshold = self._first_not_none(
+        self.group_alerts_threshold = cast(int, self._first_not_none(
             group_alerts_threshold,
             slack_config.get("group_alerts_threshold"),
             self.DEFAULT_GROUP_ALERTS_THRESHOLD,
-        )
+        ))
 
         teams_config = config.get(self._TEAMS, {})
         self.teams_webhook = self._first_not_none(
@@ -223,11 +223,13 @@ def has_send_report_platform(self):
 
     @property
     def has_slack(self) -> bool:
-        return self.slack_webhook or (self.slack_token and self.slack_channel_name)
+        return self.slack_webhook is not None or (
+            self.slack_token is not None and self.slack_channel_name is not None
+        )
 
     @property
     def has_teams(self) -> bool:
-        return self.teams_webhook
+        return self.teams_webhook is not None
 
     @property
     def has_s3(self):

elementary/monitor/alerts/alerts_groups/__init__.py

@@ -1,9 +1,11 @@
 from .alerts_group import AlertsGroup
 from .base_alerts_group import BaseAlertsGroup
+from .grouped_by_owner import GroupedByOwnerAlerts
 from .grouped_by_table import GroupedByTableAlerts
 
 __all__ = [
     "AlertsGroup",
     "BaseAlertsGroup",
     "GroupedByTableAlerts",
+    "GroupedByOwnerAlerts",
 ]

elementary/monitor/alerts/alerts_groups/grouped_by_owner.py

@@ -0,0 +1,50 @@
+from typing import Dict, List, Optional, Union
+
+from elementary.monitor.alerts.alerts_groups.alerts_group import AlertsGroup
+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.data_monitoring.alerts.integrations.utils.report_link import (
+    ReportLinkData,
+    get_test_runs_by_owner_link,
+)
+
+
+class GroupedByOwnerAlerts(AlertsGroup):
+    owner: Optional[str]
+
+    def __init__(
+        self,
+        owner: Optional[str],
+        alerts: List[Union[TestAlertModel, ModelAlertModel, SourceFreshnessAlertModel]],
+    ) -> None:
+        super().__init__(alerts)
+        self.owner = owner
+
+    @property
+    def report_url(self) -> Optional[str]:
+        return self.alerts[0].report_url
+
+    @property
+    def summary(self) -> str:
+        return f"{self.owner}: {len(self.alerts)} issues detected"
+
+    def get_report_link(self) -> Optional[ReportLinkData]:
+        if not self.model_errors:
+            return get_test_runs_by_owner_link(self.report_url, self.owner)
+
+        return None
+
+    @property
+    def unified_meta(self) -> Dict:
+        model_unified_meta = {}
+        test_unified_meta = {}
+        for alert in self.alerts:
+            alert_unified_meta = alert.unified_meta
+            if alert_unified_meta:
+                if isinstance(alert, ModelAlertModel):
+                    model_unified_meta = alert_unified_meta
+                    break
+
+                test_unified_meta = alert_unified_meta
+        return model_unified_meta or test_unified_meta

elementary/monitor/alerts/grouping_type.py

@@ -4,3 +4,4 @@
 class GroupingType(str, Enum):
     BY_ALERT = "alert"
     BY_TABLE = "table"
+    BY_OWNER = "owner"

elementary/monitor/cli.py

@@ -3,6 +3,7 @@
 import click
 
 from elementary.config.config import Config
+from elementary.monitor.alerts.grouping_type import GroupingType
 from elementary.monitor.data_monitoring.alerts.data_monitoring_alerts import (
     DataMonitoringAlerts,
 )
@@ -241,9 +242,9 @@ def get_cli_properties() -> dict:
 )
 @click.option(
     "--group-by",
-    type=click.Choice(["alert", "table"]),
+    type=click.Choice(list(map(lambda e: e.value, GroupingType))),
     default=None,
-    help="Whether to group alerts by 'alert' or by 'table'",
+    help=f"Whether to group alerts by: {', '.join(map(lambda e: e.value, GroupingType))}",
 )
 @click.option(
     "--override-dbt-project-config",

elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py

@@ -1,6 +1,6 @@
 import json
 from collections import defaultdict
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import DefaultDict, Dict, List, Optional, Union
 
 from alive_progress import alive_bar
@@ -17,7 +17,10 @@
     MessagingIntegrationError,
 )
 from elementary.monitor.alerts.alert_messages.builder import AlertMessageBuilder
-from elementary.monitor.alerts.alerts_groups import GroupedByTableAlerts
+from elementary.monitor.alerts.alerts_groups import (
+    GroupedByOwnerAlerts,
+    GroupedByTableAlerts,
+)
 from elementary.monitor.alerts.alerts_groups.alerts_group import AlertsGroup
 from elementary.monitor.alerts.grouping_type import GroupingType
 from elementary.monitor.alerts.model_alert import ModelAlertModel
@@ -159,7 +162,7 @@ def _get_suppressed_alerts(
         alerts_last_sent_times: Dict[str, datetime],
     ) -> List[str]:
         suppressed_alerts = []
-        current_time_utc = convert_time_to_timezone(datetime.utcnow())
+        current_time_utc = convert_time_to_timezone(datetime.now(timezone.utc))
         for alert in alerts:
             alert_class_id = alert.alert_class_id
             suppression_interval = alert.data.get_suppression_interval(
@@ -221,6 +224,7 @@ def _format_alerts(
         formatted_alerts = []
         grouped_by_table_alerts = []
         model_ids_to_alerts_map = defaultdict(lambda: [])
+        owner_to_alerts_map = defaultdict(lambda: [])
 
         default_alerts_group_by_strategy = GroupingType(
             self.config.slack_group_alerts_by
@@ -248,6 +252,9 @@ def _format_alerts(
                     model_ids_to_alerts_map[formatted_alert.model_unique_id].append(
                         formatted_alert
                     )
+                elif grouping_type == GroupingType.BY_OWNER:
+                    for owner in formatted_alert.owners:
+                        owner_to_alerts_map[owner].append(formatted_alert)
                 else:
                     formatted_alerts.append(formatted_alert)
             except ValueError:
@@ -266,7 +273,10 @@ def _format_alerts(
                         alerts=alerts_by_model, env=self.config.specified_env
                     )
                 )
-
+            for owner, alerts_by_owner in owner_to_alerts_map.items():
+                grouped_by_table_alerts.append(
+                    GroupedByOwnerAlerts(owner=owner, alerts=alerts_by_owner)
+                )
             self.execution_properties["had_group_by_table"] = (
                 len(grouped_by_table_alerts) > 0
             )

elementary/monitor/data_monitoring/alerts/integrations/README.md

@@ -122,7 +122,16 @@ The different alert types are:
 - Owners
 - Subscribers
 
-#### 6. Fallback template (_method name:_ `_get_fallback_template`)
+##### 6. Alerts grouped by owner (_method name:_ `_get_group_by_owner_template`)
+
+- link to report (filter by owner)
+- Owner
+- List of failures, using `alert.summary` to include model and test name.
+- List of warnings, using `alert.summary` to include model and test name.
+- Tags
+- Subscribers
+
+#### 7. Fallback template (_method name:_ `_get_fallback_template`)
 
 We try to send the formatted message and in case it fails (due to a bug or API change) we send the fallback alert, which is usually a raw JSON of the alert object.
 You can find an example of this in the Slack integration (`elementary/monitor/data_monitoring/alerts/integrations/slack.py`).

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

@@ -3,6 +3,9 @@
 
 from elementary.monitor.alerts.alerts_groups import AlertsGroup, GroupedByTableAlerts
 from elementary.monitor.alerts.alerts_groups.base_alerts_group import BaseAlertsGroup
+from elementary.monitor.alerts.alerts_groups.grouped_by_owner import (
+    GroupedByOwnerAlerts,
+)
 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
@@ -45,6 +48,8 @@ def _get_alert_template(
             return self._get_source_freshness_template(alert)
         elif isinstance(alert, GroupedByTableAlerts):
             return self._get_group_by_table_template(alert)
+        elif isinstance(alert, GroupedByOwnerAlerts):
+            return self._get_group_by_owner_template(alert)
         elif isinstance(alert, BaseAlertsGroup):
             return self._get_alerts_group_template(alert)
 
@@ -76,6 +81,12 @@ def _get_group_by_table_template(
     ):
         raise NotImplementedError
 
+    @abstractmethod
+    def _get_group_by_owner_template(
+        self, alert: GroupedByOwnerAlerts, *args, **kwargs
+    ):
+        raise NotImplementedError
+
     @abstractmethod
     def _get_alerts_group_template(self, alert: BaseAlertsGroup, *args, **kwargs):
         raise NotImplementedError

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

@@ -49,7 +49,8 @@ def add_preview_to_slack_alert(
     ):
         if preview_blocks:
             validated_preview_blocks = self._validate_preview_blocks(preview_blocks)
-            self._add_blocks_as_attachments(validated_preview_blocks)
+            if validated_preview_blocks:
+                self._add_blocks_as_attachments(validated_preview_blocks)
 
     def add_details_to_slack_alert(
         self,
@@ -86,6 +87,6 @@ def _validate_preview_blocks(cls, preview_blocks: Optional[SlackBlocksType] = No
         padding_length = (
             SlackMessageBuilder._MAX_ALERT_PREVIEW_BLOCKS - preview_blocks_count
         )
-        padding = [cls.create_empty_section_block() for i in range(padding_length)]
+        padding = [cls.create_empty_section_block() for _ in range(padding_length)]
         padded_preview_blocks.extend(padding)
         return padded_preview_blocks