feat(data_classes): Add CloudWatchAlarmEvent data class

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -7,6 +7,12 @@
 from .appsync_resolver_event import AppSyncResolverEvent
 from .aws_config_rule_event import AWSConfigRuleEvent
 from .bedrock_agent_event import BedrockAgentEvent
+from .cloud_watch_alarm_event import (
+    CloudWatchAlarmEvent,
+    CloudWatchAlarmMetric,
+    CloudWatchAlarmState,
+    CloudWatchAlarmStateValue,
+)
 from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent
 from .code_pipeline_job_event import CodePipelineJobEvent
@@ -42,6 +48,10 @@
     "AppSyncResolverEvent",
     "ALBEvent",
     "BedrockAgentEvent",
+    "CloudWatchAlarmEvent",
+    "CloudWatchAlarmMetric",
+    "CloudWatchAlarmState",
+    "CloudWatchAlarmStateValue",
     "CloudWatchDashboardCustomWidgetEvent",
     "CloudWatchLogsEvent",
     "CodePipelineJobEvent",

aws_lambda_powertools/utilities/data_classes/cloud_watch_alarm_event.py

@@ -0,0 +1,206 @@
+from __future__ import annotations
+
+import json
+from enum import Enum, auto
+from typing import List, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class CloudWatchAlarmStateValue(Enum):
+    OK = auto()
+    ALARM = auto()
+    INSUFFICIENT_DATA = auto()
+
+
+class CloudWatchAlarmState(DictWrapper):
+    @property
+    def value(self) -> CloudWatchAlarmStateValue:
+        """
+        Overall state of the alarm.
+        """
+        return CloudWatchAlarmStateValue[self["value"]]
+
+    @property
+    def reason(self) -> Optional[str]:
+        """
+        Reason why alarm was changed to this state.
+        """
+        return self.get("reason")
+
+    @property
+    def reason_data(self) -> Optional[dict]:
+        """
+        Additional data to back up the reason, usually contains the evaluated data points,
+        the calculated threshold and timestamps.
+        """
+        if self.get("reasonData") is None:
+            return None
+
+        return json.loads(str(self.get("reasonData")))
+
+    @property
+    def timestamp(self) -> str:
+        """
+        Timestamp of this state change in ISO-8601 format.
+        """
+        return self["timestamp"]
+
+
+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:
+        """
+        Unique ID of the alarm metric.
+        """
+        return self["id"]
+
+    @property
+    def expression(self) -> Optional[str]:
+        """
+        The mathematical expression for calculating the metric, if applicable.
+        """
+        return self.get("expression", None)
+
+    @property
+    def label(self) -> Optional[str]:
+        """
+        Optional label of the 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]:
+        """
+        Name of the correspondent CloudWatch Metric.
+        """
+        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.
+        """
+        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.
+        """
+        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.
+        """
+        if self._metric_stat is not None:
+            return self._metric_stat.get("stat", None)
+
+        return None
+
+    @property
+    def return_data(self) -> bool:
+        """
+        Whether this metric data is used to determine the state of the alarm or not.
+        """
+        return self["returnData"]
+
+
+class CloudWatchAlarmEvent(DictWrapper):
+    @property
+    def source(self) -> str:
+        """
+        Source of the triggered event, usually it is "aws.cloudwatch".
+        """
+        return self["source"]
+
+    @property
+    def alarm_arn(self) -> str:
+        """
+        The ARN of the CloudWatch Alarm.
+        """
+        return self["alarmArn"]
+
+    @property
+    def region(self) -> str:
+        """
+        The AWS region in which the Alarm is active.
+        """
+        return self["region"]
+
+    @property
+    def source_account_id(self) -> str:
+        """
+        The AWS Account ID that the Alarm is deployed to.
+        """
+        return self["accountId"]
+
+    @property
+    def timestamp(self) -> str:
+        """
+        Alarm state change event timestamp in ISO-8601 format.
+        """
+        return self["time"]
+
+    @property
+    def alarm_name(self) -> str:
+        """
+        Alarm name.
+        """
+        return self["alarmData"]["alarmName"]
+
+    @property
+    def alarm_description(self) -> Optional[str]:
+        """
+        Optional description for the Alarm.
+        """
+        return self["alarmData"].get("configuration", {}).get("description", None)
+
+    @property
+    def state(self):
+        """
+        The current state of the Alarm.
+        """
+        return CloudWatchAlarmState(self.get("alarmData").get("state"))
+
+    @property
+    def previous_state(self):
+        """
+        The previous state of the Alarm.
+        """
+        return CloudWatchAlarmState(self.get("alarmData").get("previousState"))
+
+    @property
+    def alarm_metrics(self) -> Optional[List[CloudWatchAlarmMetric]]:
+        maybe_metrics = self["alarmData"].get("configuration", {}).get("metrics", None)
+
+        if maybe_metrics is not None:
+            return [CloudWatchAlarmMetric(i) for i in maybe_metrics]
+
+        return None

docs/utilities/data_classes.md

@@ -86,6 +86,7 @@ Log Data Event for Troubleshooting
 | [AppSync Resolver](#appsync-resolver)                                     | `AppSyncResolverEvent`                             |
 | [AWS Config Rule](#aws-config-rule)                                       | `AWSConfigRuleEvent`                               |
 | [Bedrock Agent](#bedrock-agent)                                           | `BedrockAgent`                                     |
+| [CloudWatch Alarm State Change Action](#cloudwatch-alarm-state-change-action) | `CloudWatchAlarmEvent` |
 | [CloudWatch Dashboard Custom Widget](#cloudwatch-dashboard-custom-widget) | `CloudWatchDashboardCustomWidgetEvent`             |
 | [CloudWatch Logs](#cloudwatch-logs)                                       | `CloudWatchLogsEvent`                              |
 | [CodePipeline Job Event](#codepipeline-job)                               | `CodePipelineJobEvent`                             |
@@ -528,6 +529,23 @@ In this example, we also use the new Logger `correlation_id` and built-in `corre
         return { "markdown": f"# {echo}" }
     ```
 
+### CloudWatch Alarm State Change Action
+
+[CloudWatch supports Lambda as an alarm state change action](https://aws.amazon.com/about-aws/whats-new/2023/12/amazon-cloudwatch-alarms-lambda-change-action/){target="_blank"}.
+You can use the `CloudWathAlarmEvent` data class to access the fields containing such data as alarm information, current state, and previous state.
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.data_classes import event_source, CloudWatchAlarmEvent
+
+    @event_source(data_class=CloudWatchAlarmEvent)
+    def lambda_handler(event: CloudWatchAlarmEvent, context):
+        if event.state.value.name == "ALARM":
+            print(f"{event.alarm_name} is on alarm because {event.state.reason}...")
+            do_something_with(event.alarm_arn)
+    ```
+
 ### CloudWatch Logs
 
 CloudWatch Logs events by default are compressed and base64 encoded. You can use the helper function provided to decode,

tests/events/cloudWatchAlarmEvent.json

@@ -0,0 +1,59 @@
+{
+  "source": "aws.cloudwatch",
+  "alarmArn": "arn:aws:cloudwatch:eu-west-1:912397435824:alarm:test_alarm",
+  "accountId": "000000000000",
+  "time": "2024-02-17T11:53:08.431+0000",
+  "region": "eu-west-1",
+  "alarmData": {
+    "alarmName": "Test alert",
+    "state": {
+      "value": "ALARM",
+      "reason": "Threshold Crossed: 1 out of the last 1 datapoints [1.0 (17/02/24 11:51:00)] was less than the threshold (10.0) (minimum 1 datapoint for OK -> ALARM transition).",
+      "reasonData": "{\"version\":\"1.0\",\"queryDate\":\"2024-02-17T11:53:08.423+0000\",\"startDate\":\"2024-02-17T11:51:00.000+0000\",\"statistic\":\"SampleCount\",\"period\":60,\"recentDatapoints\":[1.0],\"threshold\":10.0,\"evaluatedDatapoints\":[{\"timestamp\":\"2024-02-17T11:51:00.000+0000\",\"sampleCount\":1.0,\"value\":1.0}]}",
+      "timestamp": "2024-02-17T11:53:08.431+0000"
+    },
+    "previousState": {
+      "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"
+    },
+    "configuration": {
+      "description": "This is description **here**",
+      "metrics": [
+        {
+          "id": "e1",
+          "expression": "m1/m2",
+          "label": "Expression1",
+          "returnData": true
+        },
+        {
+          "id": "m1",
+          "metricStat": {
+            "metric": {
+              "namespace": "AWS/Lambda",
+              "name": "Invocations",
+              "dimensions": {}
+            },
+            "period": 60,
+            "stat": "SampleCount"
+          },
+          "returnData": false
+        },
+        {
+          "id": "m2",
+          "metricStat": {
+            "metric": {
+              "namespace": "AWS/Lambda",
+              "name": "Duration",
+              "dimensions": {}
+            },
+            "period": 60,
+            "stat": "SampleCount"
+          },
+          "returnData": false
+        }
+      ]
+    }
+  }
+}

tests/unit/data_classes/test_cloud_watch_alarm_event.py

@@ -0,0 +1,58 @@
+import json
+
+from aws_lambda_powertools.utilities.data_classes.cloud_watch_alarm_event import (
+    CloudWatchAlarmEvent,
+    CloudWatchAlarmStateValue,
+)
+from tests.functional.utils import load_event
+
+
+def test_cloud_watch_alarm_event():
+    raw_event = load_event("cloudWatchAlarmEvent.json")
+    parsed_event = CloudWatchAlarmEvent(raw_event)
+
+    assert parsed_event.source == raw_event["source"]
+    assert parsed_event.region == raw_event["region"]
+    assert parsed_event.alarm_arn == raw_event["alarmArn"]
+    assert parsed_event.alarm_description == raw_event["alarmData"]["configuration"]["description"]
+    assert parsed_event.alarm_name == raw_event["alarmData"]["alarmName"]
+
+    assert parsed_event.state.value == CloudWatchAlarmStateValue[raw_event["alarmData"]["state"]["value"]]
+    assert parsed_event.state.reason == raw_event["alarmData"]["state"]["reason"]
+    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.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"]
+
+    # test the 'expression' metric
+    assert parsed_event.alarm_metrics[0].metric_id == raw_event["alarmData"]["configuration"]["metrics"][0]["id"]
+    assert (
+        parsed_event.alarm_metrics[0].expression == raw_event["alarmData"]["configuration"]["metrics"][0]["expression"]
+    )
+    assert parsed_event.alarm_metrics[0].label == raw_event["alarmData"]["configuration"]["metrics"][0]["label"]
+    assert (
+        parsed_event.alarm_metrics[0].return_data == raw_event["alarmData"]["configuration"]["metrics"][0]["returnData"]
+    )
+
+    # test the 'metric' metric
+    assert parsed_event.alarm_metrics[1].metric_id == raw_event["alarmData"]["configuration"]["metrics"][1]["id"]
+    assert (
+        parsed_event.alarm_metrics[1].name
+        == raw_event["alarmData"]["configuration"]["metrics"][1]["metricStat"]["metric"]["name"]
+    )
+    assert (
+        parsed_event.alarm_metrics[1].namespace
+        == raw_event["alarmData"]["configuration"]["metrics"][1]["metricStat"]["metric"]["namespace"]
+    )
+    assert (
+        parsed_event.alarm_metrics[1].dimensions
+        == raw_event["alarmData"]["configuration"]["metrics"][1]["metricStat"]["metric"]["dimensions"]
+    )
+    assert (
+        parsed_event.alarm_metrics[1].return_data == raw_event["alarmData"]["configuration"]["metrics"][1]["returnData"]
+    )