feat(event_sources): add CloudWatch dashboard custom widget event

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -5,6 +5,7 @@
 from .alb_event import ALBEvent
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
 from .appsync_resolver_event import AppSyncResolverEvent
+from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent
 from .code_pipeline_job_event import CodePipelineJobEvent
 from .connect_contact_flow_event import ConnectContactFlowEvent
@@ -23,6 +24,7 @@
     "APIGatewayProxyEventV2",
     "AppSyncResolverEvent",
     "ALBEvent",
+    "CloudWatchDashboardCustomWidgetEvent",
     "CloudWatchLogsEvent",
     "CodePipelineJobEvent",
     "ConnectContactFlowEvent",

aws_lambda_powertools/utilities/data_classes/cloud_watch_custom_widget_event.py

@@ -0,0 +1,153 @@
+from typing import Any, Dict, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class TimeZone(DictWrapper):
+    @property
+    def label(self) -> str:
+        """The time range label. Either 'UTC' or 'Local'"""
+        return self["label"]
+
+    @property
+    def offset_iso(self) -> str:
+        """The time range offset in the format +/-00:00"""
+        return self["offsetISO"]
+
+    @property
+    def offset_in_minutes(self) -> int:
+        """The time range offset in minutes"""
+        return int(self["offsetInMinutes"])
+
+
+class TimeRange(DictWrapper):
+    @property
+    def mode(self) -> str:
+        """The time range mode, i.e. 'relative' or 'absolute'"""
+        return self["mode"]
+
+    @property
+    def start(self) -> int:
+        """The start time within the time range"""
+        return self["start"]
+
+    @property
+    def end(self) -> int:
+        """The end time within the time range"""
+        return self["end"]
+
+    @property
+    def relative_start(self) -> Optional[int]:
+        """The relative start time within the time range"""
+        return self.get("relativeStart")
+
+    @property
+    def zoom_start(self) -> Optional[int]:
+        """The start time within the zoomed time range"""
+        return (self.get("zoom") or {}).get("start")
+
+    @property
+    def zoom_end(self) -> Optional[int]:
+        """The end time within the zoomed time range"""
+        return (self.get("zoom") or {}).get("end")
+
+
+class CloudWatchWidgetContext(DictWrapper):
+    @property
+    def dashboard_name(self) -> str:
+        """Get dashboard name, in which the widget is used"""
+        return self["dashboardName"]
+
+    @property
+    def widget_id(self) -> str:
+        """Get widget ID"""
+        return self["widgetId"]
+
+    @property
+    def account_id(self) -> str:
+        """Get AWS Account ID"""
+        return self["accountId"]
+
+    @property
+    def locale(self) -> str:
+        """Get locale language"""
+        return self["locale"]
+
+    @property
+    def timezone(self) -> TimeZone:
+        """Timezone information of the dashboard"""
+        return TimeZone(self["timezone"])
+
+    @property
+    def period(self) -> int:
+        """The period shown on the dashboard"""
+        return int(self["period"])
+
+    @property
+    def is_auto_period(self) -> bool:
+        """Whether auto period is enabled"""
+        return bool(self["isAutoPeriod"])
+
+    @property
+    def time_range(self) -> TimeRange:
+        """The widget time range"""
+        return TimeRange(self["timeRange"])
+
+    @property
+    def theme(self) -> str:
+        """The dashboard theme, i.e. 'light' or 'dark'"""
+        return self["theme"]
+
+    @property
+    def link_charts(self) -> bool:
+        """The widget is linked to other charts"""
+        return bool(self["linkCharts"])
+
+    @property
+    def title(self) -> str:
+        """Get widget title"""
+        return self["title"]
+
+    @property
+    def params(self) -> Dict[str, Any]:
+        """Get widget parameters"""
+        return self["params"]
+
+    @property
+    def forms(self) -> Dict[str, Any]:
+        """Get widget form data"""
+        return self["forms"]["all"]
+
+    @property
+    def height(self) -> int:
+        """Get widget height"""
+        return int(self["height"])
+
+    @property
+    def width(self) -> int:
+        """Get widget width"""
+        return int(self["width"])
+
+
+class CloudWatchDashboardCustomWidgetEvent(DictWrapper):
+    """CloudWatch dashboard custom widget event
+
+    You can use a Lambda function to create a custom widget on a CloudWatch dashboard.
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/add_custom_widget_dashboard_about.html
+    """
+
+    @property
+    def describe(self) -> bool:
+        """Display widget documentation"""
+        return bool(self.get("describe", False))
+
+    @property
+    def widget_context(self) -> Optional[CloudWatchWidgetContext]:
+        """The widget context"""
+        if self.get("widgetContext"):
+            return CloudWatchWidgetContext(self["widgetContext"])
+
+        return None

docs/utilities/data_classes.md

@@ -68,6 +68,7 @@ Event Source | Data_class
 [Application Load Balancer](#application-load-balancer) | `ALBEvent`
 [AppSync Authorizer](#appsync-authorizer) | `AppSyncAuthorizerEvent`
 [AppSync Resolver](#appsync-resolver) | `AppSyncResolverEvent`
+[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`
@@ -441,6 +442,41 @@ In this example, we also use the new Logger `correlation_id` and built-in `corre
     }
     ```
 
+### CloudWatch Dashboard Custom Widget
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.data_classes import event_source, CloudWatchDashboardCustomWidgetEvent
+    from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import CloudWatchLogsDecodedData
+
+    const DOCS = `
+    ## Echo
+    A simple echo script. Anything passed in \`\`\`echo\`\`\` parameter is returned as the content of custom widget.
+
+    ### Widget parameters
+    Param | Description
+    ---|---
+    **echo** | The content to echo back
+
+    ### Example parameters
+    \`\`\` yaml
+    echo: <h1>Hello world</h1>
+    \`\`\`
+    `
+
+    @event_source(data_class=CloudWatchDashboardCustomWidgetEvent)
+    def lambda_handler(event: CloudWatchDashboardCustomWidgetEvent, context):
+
+        if event.describe:
+            return DOCS
+
+        # alternatively you can also do
+        # event.widget_context.params["echo"]
+        # event.raw_event["echo"]
+        return event["echo"]
+    ```
+
 ### CloudWatch Logs
 
 CloudWatch Logs events by default are compressed and base64 encoded. You can use the helper function provided to decode,

tests/events/cloudWatchDashboardEvent.json

@@ -0,0 +1,37 @@
+{
+  "original": "param-to-widget",
+  "widgetContext": {
+    "dashboardName": "Name-of-current-dashboard",
+    "widgetId": "widget-16",
+    "accountId": "123456789123",
+    "locale": "en",
+    "timezone": {
+      "label": "UTC",
+      "offsetISO": "+00:00",
+      "offsetInMinutes": 0
+    },
+    "period": 300,
+    "isAutoPeriod": true,
+    "timeRange": {
+      "mode": "relative",
+      "start": 1627236199729,
+      "end": 1627322599729,
+      "relativeStart": 86400012,
+      "zoom": {
+        "start": 1627276030434,
+        "end": 1627282956521
+      }
+    },
+    "theme": "light",
+    "linkCharts": true,
+    "title": "Tweets for Amazon website problem",
+    "forms": {
+      "all": {}
+    },
+    "params": {
+      "original": "param-to-widget"
+    },
+    "width": 588,
+    "height": 369
+  }
+}

tests/functional/test_data_classes.py

@@ -13,6 +13,7 @@
     APIGatewayProxyEvent,
     APIGatewayProxyEventV2,
     AppSyncResolverEvent,
+    CloudWatchDashboardCustomWidgetEvent,
     CloudWatchLogsEvent,
     CodePipelineJobEvent,
     EventBridgeEvent,
@@ -99,6 +100,43 @@ def message(self) -> str:
     assert DataClassSample(data1).raw_event is data1
 
 
+def test_cloud_watch_dashboard_event():
+    event = CloudWatchDashboardCustomWidgetEvent(load_event("cloudWatchDashboardEvent.json"))
+    assert event.describe is False
+    assert event.widget_context.account_id == "123456789123"
+    assert event.widget_context.dashboard_name == "Name-of-current-dashboard"
+    assert event.widget_context.widget_id == "widget-16"
+    assert event.widget_context.locale == "en"
+    assert event.widget_context.timezone.label == "UTC"
+    assert event.widget_context.timezone.offset_iso == "+00:00"
+    assert event.widget_context.timezone.offset_in_minutes == 0
+    assert event.widget_context.period == 300
+    assert event.widget_context.is_auto_period is True
+    assert event.widget_context.time_range.mode == "relative"
+    assert event.widget_context.time_range.start == 1627236199729
+    assert event.widget_context.time_range.end == 1627322599729
+    assert event.widget_context.time_range.relative_start == 86400012
+    assert event.widget_context.time_range.zoom_start == 1627276030434
+    assert event.widget_context.time_range.zoom_end == 1627282956521
+    assert event.widget_context.theme == "light"
+    assert event.widget_context.link_charts is True
+    assert event.widget_context.title == "Tweets for Amazon website problem"
+    assert event.widget_context.forms == {}
+    assert event.widget_context.params == {"original": "param-to-widget"}
+    assert event.widget_context.width == 588
+    assert event.widget_context.height == 369
+    assert event.widget_context.params["original"] == "param-to-widget"
+    assert event["original"] == "param-to-widget"
+    assert event.raw_event["original"] == "param-to-widget"
+
+
+def test_cloud_watch_dashboard_describe_event():
+    event = CloudWatchDashboardCustomWidgetEvent({"describe": True})
+    assert event.describe is True
+    assert event.widget_context is None
+    assert event.raw_event == {"describe": True}
+
+
 def test_cloud_watch_trigger_event():
     event = CloudWatchLogsEvent(load_event("cloudWatchLogEvent.json"))