Improve annotation of the CloudWatch store (#13511)

giograno · web-flow · commit 532d755eb31e · 2025-12-16T15:39:49.000+01:00
diff --git a/localstack-core/localstack/services/cloudwatch/models.py b/localstack-core/localstack/services/cloudwatch/models.py
@@ -1,6 +1,12 @@
 import datetime
 
-from localstack.aws.api.cloudwatch import CompositeAlarm, DashboardBody, MetricAlarm, StateValue
+from localstack.aws.api.cloudwatch import (
+    AlarmHistoryItem,
+    CompositeAlarm,
+    DashboardBody,
+    MetricAlarm,
+    StateValue,
+)
 from localstack.services.stores import (
     AccountRegionBundle,
     BaseStore,
@@ -72,6 +78,8 @@ class LocalStackDashboard:
     dashboard_name: str
     dashboard_arn: str
     dashboard_body: DashboardBody
+    last_modified: datetime.datetime
+    size: int
 
     def __init__(
         self, account_id: str, region: str, dashboard_name: str, dashboard_body: DashboardBody
@@ -99,7 +107,7 @@ class CloudWatchStore(BaseStore):
 
     # Contains all the Alarm Histories. Per documentation, an alarm history is retained even if the alarm is deleted,
     # making it necessary to save this at store level
-    histories: list[dict] = LocalAttribute(default=list)
+    histories: list[dict[str, AlarmHistoryItem]] = LocalAttribute(default=list)
 
     dashboards: dict[str, LocalStackDashboard] = LocalAttribute(default=dict)
 
diff --git a/localstack-core/localstack/services/cloudwatch/provider_v2.py b/localstack-core/localstack/services/cloudwatch/provider_v2.py
@@ -9,6 +9,7 @@
 from localstack.aws.api.cloudwatch import (
     AccountId,
     ActionPrefix,
+    AlarmHistoryItem,
     AlarmName,
     AlarmNamePrefix,
     AlarmNames,
@@ -760,7 +761,7 @@ def _update_state(
         self,
         context: RequestContext,
         alarm: LocalStackAlarm,
-        state_value: str,
+        state_value: StateValue,
         state_reason: str,
         state_reason_data: dict = None,
     ):
@@ -780,18 +781,17 @@ def _update_state(
                 "stateReasonData": state_reason_data,
             },
         }
-        store.histories.append(
-            {
-                "Timestamp": timestamp_millis(alarm.alarm["StateUpdatedTimestamp"]),
-                "HistoryItemType": HistoryItemType.StateUpdate,
-                "AlarmName": alarm.alarm["AlarmName"],
-                "HistoryData": json.dumps(history_data),
-                "HistorySummary": f"Alarm updated from {old_state} to {state_value}",
-                "AlarmType": "MetricAlarm"
-                if isinstance(alarm, LocalStackMetricAlarm)
-                else "CompositeAlarm",
-            }
+        alarm_history_item = AlarmHistoryItem(
+            Timestamp=alarm.alarm["StateUpdatedTimestamp"],
+            HistoryItemType=HistoryItemType.StateUpdate,
+            AlarmName=alarm.alarm["AlarmName"],
+            HistoryData=json.dumps(history_data),
+            HistorySummary=f"Alarm updated from {old_state} to {state_value}",
+            AlarmType="MetricAlarm"
+            if isinstance(alarm, LocalStackMetricAlarm)
+            else "CompositeAlarm",
         )
+        store.histories.append(alarm_history_item)
         alarm.alarm["StateValue"] = state_value
         alarm.alarm["StateReason"] = state_reason
         if state_reason_data:
@@ -837,15 +837,10 @@ def describe_alarm_history(
         if alarm_name:
             history = [h for h in history if h["AlarmName"] == alarm_name]
 
-        def _get_timestamp(input: dict):
-            if timestamp_string := input.get("Timestamp"):
-                return datetime.datetime.fromisoformat(timestamp_string)
-            return None
-
         if start_date:
-            history = [h for h in history if (date := _get_timestamp(h)) and date >= start_date]
+            history = [h for h in history if (date := h.get("Timestamp")) and date >= start_date]
         if end_date:
-            history = [h for h in history if (date := _get_timestamp(h)) and date <= end_date]
+            history = [h for h in history if (date := h.get("Timestamp")) and date <= end_date]
         return DescribeAlarmHistoryOutput(AlarmHistoryItems=history)
 
     def _evaluate_composite_alarms(self, context: RequestContext, triggering_alarm):
