feat(event_sources): add AWS Config Rule event data class

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 .aws_config_rule_event import AWSConfigRuleEvent
 from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent
 from .code_pipeline_job_event import CodePipelineJobEvent
@@ -42,4 +43,5 @@
     "SNSEvent",
     "SQSEvent",
     "event_source",
+    "AWSConfigRuleEvent",
 ]

aws_lambda_powertools/utilities/data_classes/aws_config_rule_event.py

@@ -0,0 +1,336 @@
+from __future__ import annotations
+
+import json
+from typing import Dict, List, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+def get_invoke_event(
+    invoking_event: dict,
+) -> AWSConfigConfigurationChanged | AWSConfigScheduledNotification | AWSConfigOversizedConfiguration:
+    message_type = invoking_event.get("messageType")
+
+    if message_type == "ScheduledNotification":
+        return AWSConfigScheduledNotification(invoking_event)
+
+    if message_type == "OversizedConfigurationItemChangeNotification":
+        return AWSConfigOversizedConfiguration(invoking_event)
+
+    # Default return is ConfigurationChanged
+    return AWSConfigConfigurationChanged(invoking_event)
+
+
+class AWSConfigConfigurationChanged(DictWrapper):
+    @property
+    def configuration_item_diff(self) -> Dict:
+        """The configuration item diff of the ConfigurationItemChangeNotification event."""
+        return self["configurationItemDiff"]
+
+    @property
+    def configuration_item(self) -> AWSConfigConfigurationItemChanged:
+        """The configuration item of the ConfigurationItemChangeNotification event."""
+        return AWSConfigConfigurationItemChanged(self["configurationItem"])
+
+    @property
+    def raw_configuration_item(self) -> Dict:
+        """The raw configuration item of the ConfigurationItemChangeNotification event."""
+        return self["configurationItem"]
+
+    @property
+    def record_version(self) -> str:
+        """The record version of the ConfigurationItemChangeNotification event."""
+        return self["recordVersion"]
+
+    @property
+    def message_type(self) -> str:
+        """The message type of the ConfigurationItemChangeNotification event."""
+        return self["messageType"]
+
+    @property
+    def notification_creation_time(self) -> str:
+        """The notification creation time of the ConfigurationItemChangeNotification event."""
+        return self["notificationCreationTime"]
+
+
+class AWSConfigConfigurationItemChanged(DictWrapper):
+    @property
+    def related_events(self) -> List:
+        """The related events of the ConfigurationItemChangeNotification event."""
+        return self["relatedEvents"]
+
+    @property
+    def relationships(self) -> List:
+        """The relationships of the ConfigurationItemChangeNotification event."""
+        return self["relationships"]
+
+    @property
+    def configuration(self) -> Dict:
+        """The configuration of the ConfigurationItemChangeNotification event."""
+        return self["configuration"]
+
+    @property
+    def supplementary_configuration(self) -> Dict:
+        """The supplementary configuration of the ConfigurationItemChangeNotification event."""
+        return self["supplementaryConfiguration"]
+
+    @property
+    def tags(self) -> Dict:
+        """The tags of the ConfigurationItemChangeNotification event."""
+        return self["tags"]
+
+    @property
+    def configuration_item_version(self) -> str:
+        """The configuration item version of the ConfigurationItemChangeNotification event."""
+        return self["configurationItemVersion"]
+
+    @property
+    def configuration_item_capture_time(self) -> str:
+        """The configuration item capture time of the ConfigurationItemChangeNotification event."""
+        return self["configurationItemCaptureTime"]
+
+    @property
+    def configuration_state_id(self) -> str:
+        """The configuration state id of the ConfigurationItemChangeNotification event."""
+        return self["configurationStateId"]
+
+    @property
+    def accountid(self) -> str:
+        """The accountid of the ConfigurationItemChangeNotification event."""
+        return self["awsAccountId"]
+
+    @property
+    def configuration_item_status(self) -> str:
+        """The configuration item status of the ConfigurationItemChangeNotification event."""
+        return self["configurationItemStatus"]
+
+    @property
+    def resource_type(self) -> str:
+        """The resource type of the ConfigurationItemChangeNotification event."""
+        return self["resourceType"]
+
+    @property
+    def resource_id(self) -> str:
+        """The resource id of the ConfigurationItemChangeNotification event."""
+        return self["resourceId"]
+
+    @property
+    def resource_name(self) -> str:
+        """The resource name of the ConfigurationItemChangeNotification event."""
+        return self["resourceName"]
+
+    @property
+    def resource_arn(self) -> str:
+        """The resource arn of the ConfigurationItemChangeNotification event."""
+        return self["ARN"]
+
+    @property
+    def region(self) -> str:
+        """The region of the ConfigurationItemChangeNotification event."""
+        return self["awsRegion"]
+
+    @property
+    def availability_zone(self) -> str:
+        """The availability zone of the ConfigurationItemChangeNotification event."""
+        return self["availabilityZone"]
+
+    @property
+    def configuration_state_md5_hash(self) -> str:
+        """The md5 hash of the state of the ConfigurationItemChangeNotification event."""
+        return self["configurationStateMd5Hash"]
+
+    @property
+    def resource_creation_time(self) -> str:
+        """The resource creation time of the ConfigurationItemChangeNotification event."""
+        return self["resourceCreationTime"]
+
+
+class AWSConfigScheduledNotification(DictWrapper):
+    @property
+    def accountid(self) -> str:
+        """The accountid of the ScheduledNotification event."""
+        return self["awsAccountId"]
+
+    @property
+    def notification_creation_time(self) -> str:
+        """The notification creation time of the ScheduledNotification event."""
+        return self["notificationCreationTime"]
+
+    @property
+    def record_version(self) -> str:
+        """The record version of the ScheduledNotification event."""
+        return self["recordVersion"]
+
+    @property
+    def message_type(self) -> str:
+        """The message type of the ScheduledNotification event."""
+        return self["messageType"]
+
+
+class AWSConfigOversizedConfiguration(DictWrapper):
+    @property
+    def configuration_item_summary(self) -> AWSConfigOversizedConfigurationItemSummary:
+        """The configuration item summary of the OversizedConfiguration event."""
+        return AWSConfigOversizedConfigurationItemSummary(self["configurationItemSummary"])
+
+    @property
+    def raw_configuration_item_summary(self) -> str:
+        """The raw configuration item summary of the OversizedConfiguration event."""
+        return self["configurationItemSummary"]
+
+    @property
+    def message_type(self) -> str:
+        """The message type of the OversizedConfiguration event."""
+        return self["messageType"]
+
+    @property
+    def notification_creation_time(self) -> str:
+        """The notification creation time of the OversizedConfiguration event."""
+        return self["notificationCreationTime"]
+
+    @property
+    def record_version(self) -> str:
+        """The record version of the OversizedConfiguration event."""
+        return self["recordVersion"]
+
+
+class AWSConfigOversizedConfigurationItemSummary(DictWrapper):
+    @property
+    def change_type(self) -> str:
+        """The change type of the OversizedConfiguration event."""
+        return self["changeType"]
+
+    @property
+    def configuration_item_version(self) -> str:
+        """The configuration item version of the OversizedConfiguration event."""
+        return self["configurationItemVersion"]
+
+    @property
+    def configuration_item_capture_time(self) -> str:
+        """The configuration item capture time of the OversizedConfiguration event."""
+        return self["configurationItemCaptureTime"]
+
+    @property
+    def configuration_state_id(self) -> str:
+        """The configuration state id of the OversizedConfiguration event."""
+        return self["configurationStateId"]
+
+    @property
+    def accountid(self) -> str:
+        """The accountid of the OversizedConfiguration event."""
+        return self["awsAccountId"]
+
+    @property
+    def configuration_item_status(self) -> str:
+        """The configuration item status of the OversizedConfiguration event."""
+        return self["configurationItemStatus"]
+
+    @property
+    def resource_type(self) -> str:
+        """The resource type of the OversizedConfiguration event."""
+        return self["resourceType"]
+
+    @property
+    def resource_id(self) -> str:
+        """The resource id of the OversizedConfiguration event."""
+        return self["resourceId"]
+
+    @property
+    def resource_name(self) -> str:
+        """The resource name of the OversizedConfiguration event."""
+        return self["resourceName"]
+
+    @property
+    def resource_arn(self) -> str:
+        """The resource arn of the OversizedConfiguration event."""
+        return self["ARN"]
+
+    @property
+    def region(self) -> str:
+        """The region of the OversizedConfiguration event."""
+        return self["awsRegion"]
+
+    @property
+    def availability_zone(self) -> str:
+        """The availability zone of the OversizedConfiguration event."""
+        return self["availabilityZone"]
+
+    @property
+    def configuration_state_md5_hash(self) -> str:
+        """The state md5 hash  of the OversizedConfiguration event."""
+        return self["configurationStateMd5Hash"]
+
+    @property
+    def resource_creation_time(self) -> str:
+        """The resource creation time of the OversizedConfiguration event."""
+        return self["resourceCreationTime"]
+
+
+class AWSConfigRuleEvent(DictWrapper):
+    """Events for AWS Config Rules
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/config/latest/developerguide/evaluate-config_develop-rules_lambda-functions.html
+    """
+
+    @property
+    def version(self) -> str:
+        """The version of the event."""
+        return self["version"]
+
+    @property
+    def invoking_event(
+        self,
+    ) -> AWSConfigConfigurationChanged | AWSConfigScheduledNotification | AWSConfigOversizedConfiguration:
+        """The invoking payload of the event."""
+        return get_invoke_event(json.loads(self["invokingEvent"]))
+
+    @property
+    def raw_invoking_event(self) -> str:
+        """The raw invoking payload of the event."""
+        return self["invokingEvent"]
+
+    @property
+    def rule_parameters(self) -> Dict:
+        """The parameters of the event."""
+        return json.loads(self["ruleParameters"])
+
+    @property
+    def result_token(self) -> str:
+        """The result token of the event."""
+        return self["resultToken"]
+
+    @property
+    def event_left_scope(self) -> bool:
+        """The left scope of the event."""
+        return self["eventLeftScope"]
+
+    @property
+    def execution_role_arn(self) -> str:
+        """The execution role arn of the event."""
+        return self["executionRoleArn"]
+
+    @property
+    def config_rule_arn(self) -> str:
+        """The arn of the rule of the event."""
+        return self["configRuleArn"]
+
+    @property
+    def config_rule_name(self) -> str:
+        """The name of the rule of the event."""
+        return self["configRuleName"]
+
+    @property
+    def config_rule_id(self) -> str:
+        """The id of the rule of the event."""
+        return self["configRuleId"]
+
+    @property
+    def accountid(self) -> str:
+        """The accountid of the event."""
+        return self["accountId"]
+
+    @property
+    def evalution_mode(self) -> Optional[str]:
+        """The evalution mode of the event."""
+        return self.get("evaluationMode")

docs/utilities/data_classes.md

@@ -84,6 +84,7 @@ Log Data Event for Troubleshooting
 | [Application Load Balancer](#application-load-balancer)                   | `ALBEvent`                                         |
 | [AppSync Authorizer](#appsync-authorizer)                                 | `AppSyncAuthorizerEvent`                           |
 | [AppSync Resolver](#appsync-resolver)                                     | `AppSyncResolverEvent`                             |
+| [AWS Config Rule](#aws-config-rule)                                       | `AWSConfigRuleEvent`                               |
 | [CloudWatch Dashboard Custom Widget](#cloudwatch-dashboard-custom-widget) | `CloudWatchDashboardCustomWidgetEvent`             |
 | [CloudWatch Logs](#cloudwatch-logs)                                       | `CloudWatchLogsEvent`                              |
 | [CodePipeline Job Event](#codepipeline-job)                               | `CodePipelineJobEvent`                             |
@@ -461,6 +462,26 @@ In this example, we also use the new Logger `correlation_id` and built-in `corre
     }
     ```
 
+### AWS Config Rule
+
+=== "aws_config_rule.py"
+    ```python hl_lines="1 7"
+    --8<-- "examples/event_sources/src/aws_config_rule.py"
+    ```
+
+=== "Event - ItemChanged"
+    ```json
+    --8<-- "examples/event_sources/src/aws_config_rule_item_changed.json"
+    ```
+=== "Event - Oversized"
+    ```json
+    --8<-- "examples/event_sources/src/aws_config_rule_oversized.json"
+    ```
+=== "Event - ScheduledNotification"
+    ```json
+    --8<-- "examples/event_sources/src/aws_config_rule_scheduled.json"
+    ```
+
 ### CloudWatch Dashboard Custom Widget
 
 === "app.py"