diff --git a/aws_lambda_powertools/utilities/data_classes/__init__.py b/aws_lambda_powertools/utilities/data_classes/__init__.py index b20ec504732..20a41310523 100644 --- a/aws_lambda_powertools/utilities/data_classes/__init__.py +++ b/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", diff --git a/aws_lambda_powertools/utilities/data_classes/cloud_watch_custom_widget_event.py b/aws_lambda_powertools/utilities/data_classes/cloud_watch_custom_widget_event.py new file mode 100644 index 00000000000..40219f944ba --- /dev/null +++ b/aws_lambda_powertools/utilities/data_classes/cloud_watch_custom_widget_event.py @@ -0,0 +1,158 @@ +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 domain(self) -> str: + """AWS domain name""" + return self["domain"] + + @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 diff --git a/docs/utilities/data_classes.md b/docs/utilities/data_classes.md index a450c8788e4..86cbebd3c97 100644 --- a/docs/utilities/data_classes.md +++ b/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,40 @@ 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 + + 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:

Hello world

+ \`\`\` + ` + + @event_source(data_class=CloudWatchDashboardCustomWidgetEvent) + def lambda_handler(event: CloudWatchDashboardCustomWidgetEvent, context): + + if event.describe: + return DOCS + + # You can directly return HTML or JSON content + # Alternatively, you can return markdown that will be rendered by CloudWatch + echo = event.widget_context.params["echo"] + return { "markdown": f"# {echo}" } + ``` + ### CloudWatch Logs CloudWatch Logs events by default are compressed and base64 encoded. You can use the helper function provided to decode, diff --git a/tests/events/cloudWatchDashboardEvent.json b/tests/events/cloudWatchDashboardEvent.json new file mode 100644 index 00000000000..fd2d3be62d6 --- /dev/null +++ b/tests/events/cloudWatchDashboardEvent.json @@ -0,0 +1,38 @@ +{ + "original": "param-to-widget", + "widgetContext": { + "dashboardName": "Name-of-current-dashboard", + "widgetId": "widget-16", + "domain": "https://us-east-1.console.aws.amazon.com", + "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 + } +} diff --git a/tests/functional/test_data_classes.py b/tests/functional/test_data_classes.py index 8a87075d16c..5c7423add64 100644 --- a/tests/functional/test_data_classes.py +++ b/tests/functional/test_data_classes.py @@ -13,6 +13,7 @@ APIGatewayProxyEvent, APIGatewayProxyEventV2, AppSyncResolverEvent, + CloudWatchDashboardCustomWidgetEvent, CloudWatchLogsEvent, CodePipelineJobEvent, EventBridgeEvent, @@ -99,6 +100,44 @@ 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.domain == "https://us-east-1.console.aws.amazon.com" + 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"))