feat(data_classes): Add CloudWatchAlarmEvent data class

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -7,6 +7,14 @@
 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 (
+    CloudWatchAlarmConfiguration,
+    CloudWatchAlarmData,
+    CloudWatchAlarmEvent,
+    CloudWatchAlarmMetric,
+    CloudWatchAlarmMetricStat,
+    CloudWatchAlarmState,
+)
 from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent
 from .code_pipeline_job_event import CodePipelineJobEvent
@@ -42,6 +50,12 @@
     "AppSyncResolverEvent",
     "ALBEvent",
     "BedrockAgentEvent",
+    "CloudWatchAlarmData",
+    "CloudWatchAlarmEvent",
+    "CloudWatchAlarmMetric",
+    "CloudWatchAlarmState",
+    "CloudWatchAlarmConfiguration",
+    "CloudWatchAlarmMetricStat",
     "CloudWatchDashboardCustomWidgetEvent",
     "CloudWatchLogsEvent",
     "CodePipelineJobEvent",

aws_lambda_powertools/utilities/data_classes/cloud_watch_alarm_event.py

@@ -0,0 +1,243 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import Any, Dict, List, Literal, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class CloudWatchAlarmState(DictWrapper):
+    @property
+    def value(self) -> Literal["OK", "ALARM", "INSUFFICIENT_DATA"]:
+        """
+        Overall state of the alarm.
+        """
+        return self["value"]
+
+    @property
+    def reason(self) -> str:
+        """
+        Reason why alarm was changed to this state.
+        """
+        return self["reason"]
+
+    @property
+    def reason_data(self) -> str:
+        """
+        Additional data to back up the reason, usually contains the evaluated data points,
+        the calculated threshold and timestamps.
+        """
+        return self["reasonData"]
+
+    @cached_property
+    def reason_data_decoded(self) -> Optional[Any]:
+        """
+        Deserialized version of reason_data.
+        """
+
+        return self._json_deserializer(self.reason_data) if self.reason_data else None
+
+    @property
+    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.
+        """
+        return self.get("actionsSuppressedBy", None)
+
+    @property
+    def actions_suppressed_reason(self) -> Optional[str]:
+        """
+        Captures the reason for action suppression.
+        """
+        return self.get("actionsSuppressedReason", None)
+
+    @property
+    def timestamp(self) -> str:
+        """
+        Timestamp of this state change in ISO-8601 format.
+        """
+        return self["timestamp"]
+
+
+class CloudWatchAlarmMetric(DictWrapper):
+    @property
+    def metric_id(self) -> str:
+        """
+        Unique ID of the alarm metric.
+        """
+        return self["id"]
+
+    @property
+    def expression(self) -> Optional[str]:
+        """
+        Optional expression of the alarm metric.
+        """
+        return self.get("expression", None)
+
+    @property
+    def label(self) -> Optional[str]:
+        """
+        Optional label of the alarm metric.
+        """
+        return self.get("label", 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"]
+
+    @property
+    def metric_stat(self) -> CloudWatchAlarmMetricStat:
+        return CloudWatchAlarmMetricStat(self["metricStat"])
+
+
+class CloudWatchAlarmMetricStat(DictWrapper):
+    @property
+    def period(self) -> Optional[int]:
+        """
+        Metric evaluation period, in seconds.
+        """
+        return self.get("period", None)
+
+    @property
+    def stat(self) -> Optional[str]:
+        """
+        Statistical aggregation of metric points, e.g. Average, SampleCount, etc.
+        """
+        return self.get("stat", None)
+
+    @property
+    def unit(self) -> Optional[str]:
+        """
+        Unit for metric.
+        """
+        return self.get("unit", None)
+
+    @property
+    def metric(self) -> Optional[Dict]:
+        """
+        Metric details
+        """
+        return self.get("metric", {})
+
+
+class CloudWatchAlarmData(DictWrapper):
+    @property
+    def alarm_name(self) -> str:
+        """
+        Alarm name.
+        """
+        return self["alarmName"]
+
+    @property
+    def state(self) -> CloudWatchAlarmState:
+        """
+        The current state of the Alarm.
+        """
+        return CloudWatchAlarmState(self["state"])
+
+    @property
+    def previous_state(self) -> CloudWatchAlarmState:
+        """
+        The previous state of the Alarm.
+        """
+        return CloudWatchAlarmState(self["previousState"])
+
+    @property
+    def configuration(self) -> CloudWatchAlarmConfiguration:
+        """
+        The configuration of the Alarm.
+        """
+        return CloudWatchAlarmConfiguration(self["configuration"])
+
+
+class CloudWatchAlarmConfiguration(DictWrapper):
+    @property
+    def description(self) -> Optional[str]:
+        """
+        Optional description for the Alarm.
+        """
+        return self.get("description", 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):
+    @property
+    def source(self) -> Literal["aws.cloudwatch"]:
+        """
+        Source of the triggered event.
+        """
+        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_data(self) -> CloudWatchAlarmData:
+        """
+        Contains basic data about the Alarm and its current and previous states.
+        """
+        return CloudWatchAlarmData(self["alarmData"])

docs/utilities/data_classes.md

@@ -74,39 +74,40 @@ Log Data Event for Troubleshooting
 
 ## Supported event sources
 
-| Event Source                                                              | Data_class                                         |
-|---------------------------------------------------------------------------|----------------------------------------------------|
-| [Active MQ](#active-mq)                                                   | `ActiveMQEvent`                                    |
-| [API Gateway Authorizer](#api-gateway-authorizer)                         | `APIGatewayAuthorizerRequestEvent`                 |
-| [API Gateway Authorizer V2](#api-gateway-authorizer-v2)                   | `APIGatewayAuthorizerEventV2`                      |
-| [API Gateway Proxy](#api-gateway-proxy)                                   | `APIGatewayProxyEvent`                             |
-| [API Gateway Proxy V2](#api-gateway-proxy-v2)                             | `APIGatewayProxyEventV2`                           |
-| [Application Load Balancer](#application-load-balancer)                   | `ALBEvent`                                         |
-| [AppSync Authorizer](#appsync-authorizer)                                 | `AppSyncAuthorizerEvent`                           |
-| [AppSync Resolver](#appsync-resolver)                                     | `AppSyncResolverEvent`                             |
-| [AWS Config Rule](#aws-config-rule)                                       | `AWSConfigRuleEvent`                               |
-| [Bedrock Agent](#bedrock-agent)                                           | `BedrockAgent`                                     |
-| [CloudWatch Dashboard Custom Widget](#cloudwatch-dashboard-custom-widget) | `CloudWatchDashboardCustomWidgetEvent`             |
-| [CloudWatch Logs](#cloudwatch-logs)                                       | `CloudWatchLogsEvent`                              |
-| [CodePipeline Job Event](#codepipeline-job)                               | `CodePipelineJobEvent`                             |
-| [Cognito User Pool](#cognito-user-pool)                                   | Multiple available under `cognito_user_pool_event` |
-| [Connect Contact Flow](#connect-contact-flow)                             | `ConnectContactFlowEvent`                          |
-| [DynamoDB streams](#dynamodb-streams)                                     | `DynamoDBStreamEvent`, `DynamoDBRecordEventName`   |
-| [EventBridge](#eventbridge)                                               | `EventBridgeEvent`                                 |
-| [Kafka](#kafka)                                                           | `KafkaEvent`                                       |
-| [Kinesis Data Stream](#kinesis-streams)                                   | `KinesisStreamEvent`                               |
-| [Kinesis Firehose Delivery Stream](#kinesis-firehose-delivery-stream)     | `KinesisFirehoseEvent`                             |
-| [Lambda Function URL](#lambda-function-url)                               | `LambdaFunctionUrlEvent`                           |
-| [Rabbit MQ](#rabbit-mq)                                                   | `RabbitMQEvent`                                    |
-| [S3](#s3)                                                                 | `S3Event`                                          |
-| [S3 Batch Operations](#s3-batch-operations)                               | `S3BatchOperationEvent`                            |
-| [S3 Object Lambda](#s3-object-lambda)                                     | `S3ObjectLambdaEvent`                              |
-| [S3 EventBridge Notification](#s3-eventbridge-notification)               | `S3EventBridgeNotificationEvent`                   |
-| [SES](#ses)                                                               | `SESEvent`                                         |
-| [SNS](#sns)                                                               | `SNSEvent`                                         |
-| [SQS](#sqs)                                                               | `SQSEvent`                                         |
-| [VPC Lattice V2](#vpc-lattice-v2)                                         | `VPCLatticeV2Event`                                |
-| [VPC Lattice V1](#vpc-lattice-v1)                                         | `VPCLatticeEvent`                                  |
+| Event Source                                                                  | Data_class                                         |
+|-------------------------------------------------------------------------------|----------------------------------------------------|
+| [Active MQ](#active-mq)                                                       | `ActiveMQEvent`                                    |
+| [API Gateway Authorizer](#api-gateway-authorizer)                             | `APIGatewayAuthorizerRequestEvent`                 |
+| [API Gateway Authorizer V2](#api-gateway-authorizer-v2)                       | `APIGatewayAuthorizerEventV2`                      |
+| [API Gateway Proxy](#api-gateway-proxy)                                       | `APIGatewayProxyEvent`                             |
+| [API Gateway Proxy V2](#api-gateway-proxy-v2)                                 | `APIGatewayProxyEventV2`                           |
+| [Application Load Balancer](#application-load-balancer)                       | `ALBEvent`                                         |
+| [AppSync Authorizer](#appsync-authorizer)                                     | `AppSyncAuthorizerEvent`                           |
+| [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`                             |
+| [Cognito User Pool](#cognito-user-pool)                                       | Multiple available under `cognito_user_pool_event` |
+| [Connect Contact Flow](#connect-contact-flow)                                 | `ConnectContactFlowEvent`                          |
+| [DynamoDB streams](#dynamodb-streams)                                         | `DynamoDBStreamEvent`, `DynamoDBRecordEventName`   |
+| [EventBridge](#eventbridge)                                                   | `EventBridgeEvent`                                 |
+| [Kafka](#kafka)                                                               | `KafkaEvent`                                       |
+| [Kinesis Data Stream](#kinesis-streams)                                       | `KinesisStreamEvent`                               |
+| [Kinesis Firehose Delivery Stream](#kinesis-firehose-delivery-stream)         | `KinesisFirehoseEvent`                             |
+| [Lambda Function URL](#lambda-function-url)                                   | `LambdaFunctionUrlEvent`                           |
+| [Rabbit MQ](#rabbit-mq)                                                       | `RabbitMQEvent`                                    |
+| [S3](#s3)                                                                     | `S3Event`                                          |
+| [S3 Batch Operations](#s3-batch-operations)                                   | `S3BatchOperationEvent`                            |
+| [S3 Object Lambda](#s3-object-lambda)                                         | `S3ObjectLambdaEvent`                              |
+| [S3 EventBridge Notification](#s3-eventbridge-notification)                   | `S3EventBridgeNotificationEvent`                   |
+| [SES](#ses)                                                                   | `SESEvent`                                         |
+| [SNS](#sns)                                                                   | `SNSEvent`                                         |
+| [SQS](#sqs)                                                                   | `SQSEvent`                                         |
+| [VPC Lattice V2](#vpc-lattice-v2)                                             | `VPCLatticeV2Event`                                |
+| [VPC Lattice V1](#vpc-lattice-v1)                                             | `VPCLatticeEvent`                                  |
 
 ???+ info
     The examples provided below are far from exhaustive - the data classes themselves are designed to provide a form of
@@ -528,6 +529,17 @@ 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://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/AlarmThatSendsEmail.html#alarms-and-actions){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 hl_lines="2 8"
+    --8<-- "examples/event_sources/src/cloudwatch_alarm_event.py"
+    ```
+
 ### CloudWatch Logs
 
 CloudWatch Logs events by default are compressed and base64 encoded. You can use the helper function provided to decode,

examples/event_sources/src/cloudwatch_alarm_event.py

@@ -0,0 +1,19 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import CloudWatchAlarmEvent, event_source
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+logger = Logger()
+
+
+@event_source(data_class=CloudWatchAlarmEvent)
+def lambda_handler(event: CloudWatchAlarmEvent, context: LambdaContext) -> dict:
+    logger.info(f"Alarm {event.alarm_data.alarm_name} state is {event.alarm_data.state.value}")
+
+    # You can now work with event. For example, you can enrich the received data, and
+    # decide on how you want to route the alarm.
+
+    return {
+        "name": event.alarm_data.alarm_name,
+        "arn": event.alarm_arn,
+        "urgent": "Priority: P1" in (event.alarm_data.configuration.description or ""),
+    }