fix(data_classes): address mypy issues

Author: par6n

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -9,9 +9,9 @@
 from .bedrock_agent_event import BedrockAgentEvent
 from .cloud_watch_alarm_event import (
     CloudWatchAlarmEvent,
-    CloudWatchAlarmStateValue,
-    CloudWatchAlarmState,
     CloudWatchAlarmMetric,
+    CloudWatchAlarmState,
+    CloudWatchAlarmStateValue,
 )
 from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent

aws_lambda_powertools/utilities/data_classes/cloud_watch_alarm_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 from enum import Enum, auto
 from typing import List, Optional
@@ -32,7 +34,10 @@ def reason_data(self) -> Optional[dict]:
         Additional data to back up the reason, usually contains the evaluated data points,
         the calculated threshold and timestamps.
         """
-        return json.loads(self.get("reasonData")) if self.get("reasonData") is not None else None
+        if self.get("reasonData") is None:
+            return None
+
+        return json.loads(str(self.get("reasonData")))
 
     @property
     def timestamp(self) -> str:
@@ -43,6 +48,11 @@ 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:
         """
@@ -55,35 +65,50 @@ def namespace(self) -> Optional[str]:
         """
         Namespace of the correspondent CloudWatch Metric.
         """
-        return self.get("metricStat", {}).get("metric", {}).get("namespace", None)
+        if self._metric_stat is not None:
+            return self._metric_stat.get("metric", {}).get("namespace", None)
+
+        return None
 
     @property
     def name(self) -> Optional[str]:
         """
         Name of the correspondent CloudWatch Metric.
         """
-        return self.get("metricStat", {}).get("metric", {}).get("name", None)
+        if self._metric_stat is not None:
+            return self._metric_stat.get("metric", {}).get("name", None)
+
+        return None
 
     @property
     def dimensions(self) -> Optional[dict]:
         """
         Additional dimensions of the correspondent CloudWatch Metric, if available.
         """
-        return self.get("metricStat", {}).get("metric", {}).get("dimensions", None)
+        if self._metric_stat is not None:
+            return self._metric_stat.get("metric", {}).get("dimensions", None)
+
+        return None
 
     @property
     def period(self) -> Optional[int]:
         """
         Metric evaluation period, in seconds.
         """
-        return self.get("metricStat", {}).get("period", None)
+        if self._metric_stat is not None:
+            return self._metric_stat.get("period", None)
+
+        return None
 
     @property
     def stat(self) -> Optional[str]:
         """
         Statistical aggregation of metric points, e.g. Average, SampleCount, etc.
         """
-        return self.get("metricStat", {}).get("stat", None)
+        if self._metric_stat is not None:
+            return self._metric_stat.get("stat", None)
+
+        return None
 
     @property
     def return_data(self) -> bool:
@@ -131,14 +156,14 @@ def alarm_name(self) -> str:
         """
         Alarm name.
         """
-        return self.get("alarmData").get("alarmName")
+        return self["alarmData"]["alarmName"]
 
     @property
     def alarm_description(self) -> Optional[str]:
         """
         Optional description for the Alarm.
         """
-        return self.get("alarmData").get("configuration", {}).get("description", None)
+        return self["alarmData"].get("configuration", {}).get("description", None)
 
     @property
     def state(self):
@@ -156,7 +181,7 @@ def previous_state(self):
 
     @property
     def alarm_metrics(self) -> Optional[List[CloudWatchAlarmMetric]]:
-        maybe_metrics = self.get("alarmData", {}).get("configuration", {}).get("metrics", None)
+        maybe_metrics = self["alarmData"].get("configuration", {}).get("metrics", None)
 
         if maybe_metrics is not None:
             return [CloudWatchAlarmMetric(i) for i in maybe_metrics]

tests/unit/data_classes/test_cloud_watch_alarm_event.py

@@ -1,8 +1,8 @@
 import json
 
 from aws_lambda_powertools.utilities.data_classes.cloud_watch_alarm_event import (
-    CloudWatchAlarmStateValue,
     CloudWatchAlarmEvent,
+    CloudWatchAlarmStateValue,
 )
 from tests.functional.utils import load_event
 
@@ -22,7 +22,9 @@ def test_cloud_watch_alarm_event():
     assert parsed_event.state.reason_data == json.loads(raw_event["alarmData"]["state"]["reasonData"])
     assert parsed_event.state.timestamp == raw_event["alarmData"]["state"]["timestamp"]
 
-    assert parsed_event.previous_state.value == CloudWatchAlarmStateValue[raw_event["alarmData"]["previousState"]["value"]]
+    assert (
+        parsed_event.previous_state.value == CloudWatchAlarmStateValue[raw_event["alarmData"]["previousState"]["value"]]
+    )
     assert parsed_event.previous_state.reason == raw_event["alarmData"]["previousState"]["reason"]
     assert parsed_event.previous_state.reason_data == json.loads(raw_event["alarmData"]["previousState"]["reasonData"])
     assert parsed_event.previous_state.timestamp == raw_event["alarmData"]["previousState"]["timestamp"]