Refactoring code

Author: leandrodamascena

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -8,9 +8,11 @@
 from .aws_config_rule_event import AWSConfigRuleEvent
 from .bedrock_agent_event import BedrockAgentEvent
 from .cloud_watch_alarm_event import (
+    CloudWatchAlarmConfiguration,
     CloudWatchAlarmData,
     CloudWatchAlarmEvent,
     CloudWatchAlarmMetric,
+    CloudWatchAlarmMetricStat,
     CloudWatchAlarmState,
 )
 from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
@@ -52,6 +54,8 @@
     "CloudWatchAlarmEvent",
     "CloudWatchAlarmMetric",
     "CloudWatchAlarmState",
+    "CloudWatchAlarmConfiguration",
+    "CloudWatchAlarmMetricStat",
     "CloudWatchDashboardCustomWidgetEvent",
     "CloudWatchLogsEvent",
     "CodePipelineJobEvent",

aws_lambda_powertools/utilities/data_classes/cloud_watch_alarm_event.py

@@ -1,31 +1,28 @@
 from __future__ import annotations
 
 from functools import cached_property
-from typing import Any, List, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
-CloudWatchAlarmStateValue = Literal["OK", "ALARM", "INSUFFICIENT_DATA"]
-CloudWatchAlarmActionSuppressor = Literal["Alarm", "ExtensionPeriod", "WaitPeriod"]
-
 
 class CloudWatchAlarmState(DictWrapper):
     @property
-    def value(self) -> CloudWatchAlarmStateValue:
+    def value(self) -> Literal["OK", "ALARM", "INSUFFICIENT_DATA"]:
         """
         Overall state of the alarm.
         """
         return self["value"]
 
     @property
-    def reason(self) -> Optional[str]:
+    def reason(self) -> str:
         """
         Reason why alarm was changed to this state.
         """
         return self.get("reason")
 
     @property
-    def reason_data(self) -> Optional[str]:
+    def reason_data(self) -> str:
         """
         Additional data to back up the reason, usually contains the evaluated data points,
         the calculated threshold and timestamps.
@@ -37,13 +34,11 @@ def reason_data_decoded(self) -> Optional[Any]:
         """
         Deserialized version of reason_data.
         """
-        if self.reason_data is None:
-            return None
 
-        return self._json_deserializer(self.reason_data)
+        return self._json_deserializer(self.reason_data) if self.reason_data else None
 
     @property
-    def actions_suppressed_by(self) -> Optional[CloudWatchAlarmActionSuppressor]:
+    def actions_suppressed_by(self) -> Optional[Literal["Alarm", "ExtensionPeriod", "WaitPeriod"]]:
         """
         Describes why the actions when the value is `ALARM` are suppressed in a composite
         alarm.
@@ -66,11 +61,6 @@ def timestamp(self) -> str:
 
 
 class CloudWatchAlarmMetric(DictWrapper):
-    def __init__(self, data: dict):
-        super().__init__(data)
-
-        self._metric_stat: dict | None = self.get("metricStat")
-
     @property
     def metric_id(self) -> str:
         """
@@ -81,98 +71,67 @@ def metric_id(self) -> str:
     @property
     def expression(self) -> Optional[str]:
         """
-        The mathematical expression for calculating the metric, if applicable.
+        Optional expression of the alarm metric.
         """
         return self.get("expression", None)
 
     @property
-    def label(self) -> Optional[str]:
+    def label(self) -> str:
         """
-        Optional label of the metric.
+        Optional label of the alarm metric.
         """
         return self.get("label", None)
 
     @property
-    def namespace(self) -> Optional[str]:
-        """
-        Namespace of the correspondent CloudWatch Metric.
-        """
-        if self._metric_stat is not None:
-            return self._metric_stat.get("metric", {}).get("namespace", None)
-
-        return None
-
-    @property
-    def name(self) -> Optional[str]:
+    def return_data(self) -> bool:
         """
-        Name of the correspondent CloudWatch Metric.
+        Whether this metric data is used to determine the state of the alarm or not.
         """
-        if self._metric_stat is not None:
-            return self._metric_stat.get("metric", {}).get("name", None)
-
-        return None
+        return self["returnData"]
 
     @property
-    def dimensions(self) -> Optional[dict]:
-        """
-        Additional dimensions of the correspondent CloudWatch Metric, if available.
-        """
-        if self._metric_stat is not None:
-            return self._metric_stat.get("metric", {}).get("dimensions", None)
+    def metric_stat(self) -> CloudWatchAlarmMetricStat:
+        return CloudWatchAlarmMetricStat(self["metricStat"])
 
-        return None
 
+class CloudWatchAlarmMetricStat(DictWrapper):
     @property
     def period(self) -> Optional[int]:
         """
         Metric evaluation period, in seconds.
         """
-        if self._metric_stat is not None:
-            return self._metric_stat.get("period", None)
-
-        return None
+        return self.get("period", None)
 
     @property
     def stat(self) -> Optional[str]:
         """
         Statistical aggregation of metric points, e.g. Average, SampleCount, etc.
         """
-        if self._metric_stat is not None:
-            return self._metric_stat.get("stat", None)
+        return self.get("stat", None)
 
-        return None
+    @property
+    def unit(self) -> Optional[str]:
+        """
+        Unit for metric.
+        """
+        return self.get("unit", None)
 
     @property
-    def return_data(self) -> bool:
+    def metric(self) -> Dict:
         """
-        Whether this metric data is used to determine the state of the alarm or not.
+        Metric details
         """
-        return self["returnData"]
+        return self.get("metric", {})
 
 
 class CloudWatchAlarmData(DictWrapper):
-    def __init__(self, data: dict):
-        super().__init__(data)
-
-        self._configuration = self.get("configuration", None)
-
     @property
-    def name(self) -> str:
+    def alarm_name(self) -> str:
         """
         Alarm name.
         """
         return self["alarmName"]
 
-    @property
-    def description(self) -> Optional[str]:
-        """
-        Optional description for the Alarm.
-        """
-        if self._configuration is not None:
-            return self._configuration.get("description", None)
-
-        return None
-
     @property
     def state(self) -> CloudWatchAlarmState:
         """
@@ -188,19 +147,56 @@ def previous_state(self) -> CloudWatchAlarmState:
         return CloudWatchAlarmState(self["previousState"])
 
     @property
-    def metrics(self) -> Optional[List[CloudWatchAlarmMetric]]:
+    def configuration(self) -> CloudWatchAlarmConfiguration:
         """
-        The metrics evaluated for the Alarm.
+        The configuration of the Alarm.
         """
-        if self._configuration is None:
-            return None
+        return CloudWatchAlarmConfiguration(self["configuration"])
 
-        maybe_metrics = self._configuration.get("metrics", None)
 
-        if maybe_metrics is not None:
-            return [CloudWatchAlarmMetric(i) for i in maybe_metrics]
+class CloudWatchAlarmConfiguration(DictWrapper):
+    @property
+    def description(self) -> Optional[str]:
+        """
+        Optional description for the Alarm.
+        """
+        return self.get("description", None)
 
-        return None
+    @property
+    def alarm_rule(self) -> Optional[str]:
+        """
+        Optional description for the Alarm rule in case of composite alarm.
+        """
+        return self.get("alarmRule", None)
+
+    @property
+    def alarm_actions_suppressor(self) -> Optional[str]:
+        """
+        Optional action suppression for the Alarm rule in case of composite alarm.
+        """
+        return self.get("actionsSuppressor", None)
+
+    @property
+    def alarm_actions_suppressor_wait_period(self) -> Optional[str]:
+        """
+        Optional action suppression wait period for the Alarm rule in case of composite alarm.
+        """
+        return self.get("actionsSuppressorWaitPeriod", None)
+
+    @property
+    def alarm_actions_suppressor_extension_period(self) -> Optional[str]:
+        """
+        Optional action suppression extension period for the Alarm rule in case of composite alarm.
+        """
+        return self.get("actionsSuppressorExtensionPeriod", None)
+
+    @property
+    def metrics(self) -> Optional[List[CloudWatchAlarmMetric]]:
+        """
+        The metrics evaluated for the Alarm.
+        """
+        metrics = self.get("metrics")
+        return [CloudWatchAlarmMetric(i) for i in metrics] if metrics else None
 
 
 class CloudWatchAlarmEvent(DictWrapper):

tests/events/cloudWatchAlarmEventCompositeMetric.json

@@ -0,0 +1,30 @@
+{
+  "source":"aws.cloudwatch",
+  "alarmArn":"arn:aws:cloudwatch:us-east-1:111122223333:alarm:SuppressionDemo.Main",
+  "accountId":"111122223333",
+  "time":"2023-08-04T12:56:46.138+0000",
+  "region":"us-east-1",
+  "alarmData":{
+     "alarmName":"CompositeDemo.Main",
+     "state":{
+        "value":"ALARM",
+        "reason":"arn:aws:cloudwatch:us-east-1:111122223333:alarm:CompositeDemo.FirstChild transitioned to ALARM at Friday 04 August, 2023 12:54:46 UTC",
+        "reasonData":"{\"triggeringAlarms\":[{\"arn\":\"arn:aws:cloudwatch:us-east-1:111122223333:alarm:CompositeDemo.FirstChild\",\"state\":{\"value\":\"ALARM\",\"timestamp\":\"2023-08-04T12:54:46.138+0000\"}}]}",
+        "timestamp":"2023-08-04T12:56:46.138+0000"
+     },
+     "previousState":{
+        "value":"ALARM",
+        "reason":"arn:aws:cloudwatch:us-east-1:111122223333:alarm:CompositeDemo.FirstChild transitioned to ALARM at Friday 04 August, 2023 12:54:46 UTC",
+        "reasonData":"{\"triggeringAlarms\":[{\"arn\":\"arn:aws:cloudwatch:us-east-1:111122223333:alarm:CompositeDemo.FirstChild\",\"state\":{\"value\":\"ALARM\",\"timestamp\":\"2023-08-04T12:54:46.138+0000\"}}]}",
+        "timestamp":"2023-08-04T12:54:46.138+0000",
+        "actionsSuppressedBy":"WaitPeriod",
+        "actionsSuppressedReason":"Actions suppressed by WaitPeriod"
+     },
+     "configuration":{
+        "alarmRule":"ALARM(CompositeDemo.FirstChild) OR ALARM(CompositeDemo.SecondChild)",
+        "actionsSuppressor":"CompositeDemo.ActionsSuppressor",
+        "actionsSuppressorWaitPeriod":120,
+        "actionsSuppressorExtensionPeriod":180
+     }
+  }
+}

tests/events/cloudWatchAlarmEvent.json renamed to tests/events/cloudWatchAlarmEventSingleMetric.json

@@ -16,9 +16,7 @@
       "value": "OK",
       "reason": "Threshold Crossed: 1 out of the last 1 datapoints [1.0 (17/02/24 11:50:00)] was not greater than the threshold (10.0) (minimum 1 datapoint for ALARM -> OK transition).",
       "reasonData": "{\"version\":\"1.0\",\"queryDate\":\"2024-02-17T11:51:31.460+0000\",\"startDate\":\"2024-02-17T11:50:00.000+0000\",\"statistic\":\"SampleCount\",\"period\":60,\"recentDatapoints\":[1.0],\"threshold\":10.0,\"evaluatedDatapoints\":[{\"timestamp\":\"2024-02-17T11:50:00.000+0000\",\"sampleCount\":1.0,\"value\":1.0}]}",
-      "timestamp": "2024-02-17T11:51:31.462+0000",
-      "actionsSuppressedBy": "WaitPeriod",
-      "actionsSuppressedReason": "Actions suppressed by WaitPeriod"
+      "timestamp": "2024-02-17T11:51:31.462+0000"
     },
     "configuration": {
       "description": "This is description **here**",
@@ -58,4 +56,4 @@
       ]
     }
   }
-}
+}