feat: add CloudFormationCustomResourceEvent event source data class.

Author: phipag

aws_lambda_powertools/logging/constants.py

@@ -1,5 +1,5 @@
 # logger.powertools_handler is set with Powertools Logger handler; useful when there are many handlers
 LOGGER_ATTRIBUTE_POWERTOOLS_HANDLER = "powertools_handler"
-# logger.init attribute is set when Logger has been configured 
+# logger.init attribute is set when Logger has been configured
 LOGGER_ATTRIBUTE_PRECONFIGURED = "init"
 LOGGER_ATTRIBUTE_HANDLER = "logger_handler"

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -17,6 +17,7 @@
 )
 from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent
+from .cloudformation_custom_resource_event import CloudFormationCustomResourceEvent
 from .code_pipeline_job_event import CodePipelineJobEvent
 from .connect_contact_flow_event import ConnectContactFlowEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
@@ -81,4 +82,5 @@
     "AWSConfigRuleEvent",
     "VPCLatticeEvent",
     "VPCLatticeEventV2",
+    "CloudFormationCustomResourceEvent",
 ]

aws_lambda_powertools/utilities/data_classes/cloudformation_custom_resource_event.py

@@ -0,0 +1,52 @@
+from enum import Enum
+from typing import Any, Dict, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class CloudFormationRequestType(Enum):
+    CREATE = "Create"
+    UPDATE = "Update"
+    DELETE = "Delete"
+
+
+class CloudFormationCustomResourceEvent(DictWrapper):
+    @property
+    def request_type(self) -> CloudFormationRequestType:
+        return CloudFormationRequestType(self["RequestType"])
+
+    @property
+    def service_token(self) -> str:
+        return self["ServiceToken"]
+
+    @property
+    def response_url(self) -> str:
+        return self["ResponseUrl"]
+
+    @property
+    def stack_id(self) -> str:
+        return self["StackId"]
+
+    @property
+    def request_id(self) -> str:
+        return self["RequestId"]
+
+    @property
+    def logical_resource_id(self) -> str:
+        return self["LogicalResourceId"]
+
+    @property
+    def physical_resource_id(self) -> Optional[str]:
+        return self.get("PhysicalResourceId")
+
+    @property
+    def resource_type(self) -> str:
+        return self["ResourceType"]
+
+    @property
+    def resource_properties(self) -> Optional[Dict[str, Any]]:
+        return self.get("ResourceProperties")
+
+    @property
+    def old_resource_properties(self) -> Optional[Dict[str, Any]]:
+        return self.get("OldResourceProperties")

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -68,7 +68,11 @@
     S3Model,
     S3RecordModel,
 )
-from .s3_batch_operation import S3BatchOperationJobModel, S3BatchOperationModel, S3BatchOperationTaskModel
+from .s3_batch_operation import (
+    S3BatchOperationJobModel,
+    S3BatchOperationModel,
+    S3BatchOperationTaskModel,
+)
 from .s3_event_notification import (
     S3SqsEventNotificationModel,
     S3SqsEventNotificationRecordModel,

aws_lambda_powertools/utilities/parser/models/cloudformation_custom_resource.py

@@ -22,8 +22,10 @@ class CloudFormationCustomResourceCreateModel(CloudFormationCustomResourceBaseMo
 
 class CloudFormationCustomResourceDeleteModel(CloudFormationCustomResourceBaseModel):
     request_type: Literal["Delete"] = Field(..., alias="RequestType")
+    physical_resource_id: str = Field(..., alias="PhysicalResourceId")
 
 
 class CloudFormationCustomResourceUpdateModel(CloudFormationCustomResourceBaseModel):
     request_type: Literal["Update"] = Field(..., alias="RequestType")
+    physical_resource_id: str = Field(..., alias="PhysicalResourceId")
     old_resource_properties: Union[Dict[str, Any], BaseModel, None] = Field(None, alias="OldResourceProperties")

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`                                     |
+| [CloudFormation Custom Resource](#cloudformation-custom-resource)             | `CloudFormationCustomResourceEvent`                |
 | [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`                              |
@@ -495,6 +496,54 @@ In this example, we also use the new Logger `correlation_id` and built-in `corre
     --8<-- "examples/event_sources/src/bedrock_agent_event.py"
     ```
 
+### CloudFormation Custom Resource
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.data_classes import event_source, CloudFormationCustomResourceEvent
+    from aws_lambda_powertools.utilities.data_classes.cloudformation_custom_resource_event import CloudFormationRequestType
+    from aws_lambda_powertools.utilities.typing import LambdaContext
+    from aws_lambda_powertools import Logger
+
+    logger = Logger()
+
+
+    @event_source(data_class=CloudFormationCustomResourceEvent)
+    def lambda_handler(event: CloudFormationCustomResourceEvent, context: LambdaContext):
+        request_type = event.request_type
+
+        if request_type == CloudFormationRequestType.CREATE:
+            return on_create(event)
+        if request_type == CloudFormationRequestType.UPDATE:
+            return on_update(event)
+        if request_type == CloudFormationRequestType.DELETE:
+            return on_delete(event)
+
+
+    def on_create(event: CloudFormationCustomResourceEvent):
+        props = event.resource_properties
+        logger.info(f"Create new resource with props {props}.")
+
+        # Add your create code here ...
+        physical_id = ...
+
+        return {"PhysicalResourceId": physical_id}
+
+
+    def on_update(event: CloudFormationCustomResourceEvent):
+        physical_id = event.physical_resource_id
+        props = event.resource_properties
+        logger.info(f"Update resource {physical_id} with props {props}.")
+        # ...
+
+
+    def on_delete(event: CloudFormationCustomResourceEvent):
+        physical_id = event.physical_resource_id
+        logger.info(f"Delete resource {physical_id}.")
+        # ...
+    ```
+
 ### CloudWatch Dashboard Custom Widget
 
 === "app.py"

tests/events/cloudformationCustomResourceDelete.json

@@ -5,9 +5,10 @@
   "StackId": "arn:aws:cloudformation:us-east-1:xxxx:stack/xxxx/271845b0-f2e8-11ed-90ac-0eeb25b8ae21",
   "RequestId": "xxxxx-d2a0-4dfb-ab1f-xxxxxx",
   "LogicalResourceId": "xxxxxxxxx",
+  "PhysicalResourceId": "xxxxxxxxx",
   "ResourceType": "Custom::MyType",
   "ResourceProperties": {
     "ServiceToken": "arn:aws:lambda:us-east-1:xxxxx:function:xxxxx",
     "MyProps": "ss"
   }
-}
+}

tests/events/cloudformationCustomResourceUpdate.json

@@ -5,6 +5,7 @@
   "StackId": "arn:aws:cloudformation:us-east-1:xxxx:stack/xxxx/271845b0-f2e8-11ed-90ac-0eeb25b8ae21",
   "RequestId": "xxxxx-d2a0-4dfb-ab1f-xxxxxx",
   "LogicalResourceId": "xxxxxxxxx",
+  "PhysicalResourceId": "xxxxxxxxx",
   "ResourceType": "Custom::MyType",
   "ResourceProperties": {
     "ServiceToken": "arn:aws:lambda:us-east-1:xxxxx:function:xxxxx",
@@ -14,4 +15,4 @@
     "ServiceToken": "arn:aws:lambda:us-east-1:xxxxx:function:xxxxx-xxxx-xxx",
     "MyProps": "old"
   }
-}
+}

tests/unit/data_classes/test_cloudformation_custom_resource_event.py

@@ -0,0 +1,32 @@
+import pytest
+
+from aws_lambda_powertools.utilities.data_classes import (
+    CloudFormationCustomResourceEvent,
+)
+from aws_lambda_powertools.utilities.data_classes.cloudformation_custom_resource_event import (
+    CloudFormationRequestType,
+)
+from tests.functional.utils import load_event
+
+
+@pytest.mark.parametrize(
+    "event_file",
+    [
+        "cloudformationCustomResourceCreate.json",
+        "cloudformationCustomResourceUpdate.json",
+        "cloudformationCustomResourceDelete.json",
+    ],
+)
+def test_cloudformation_custom_resource_event(event_file):
+    raw_event = load_event(event_file)
+    parsed_event = CloudFormationCustomResourceEvent(raw_event)
+
+    assert parsed_event.request_type == CloudFormationRequestType(raw_event["RequestType"])
+    assert isinstance(parsed_event.request_type, CloudFormationRequestType)
+    assert parsed_event.service_token == raw_event["ServiceToken"]
+    assert parsed_event.stack_id == raw_event["StackId"]
+    assert parsed_event.request_id == raw_event["RequestId"]
+    assert parsed_event.logical_resource_id == raw_event["LogicalResourceId"]
+    assert parsed_event.resource_type == raw_event["ResourceType"]
+    assert parsed_event.resource_properties == raw_event.get("ResourceProperties")
+    assert parsed_event.old_resource_properties == raw_event.get("OldResourceProperties")

tests/unit/data_classes/test_sqs_event.py

@@ -65,10 +65,7 @@ def test_sqs_dlq_trigger_event():
     assert attributes.sequence_number is None
     assert attributes.message_group_id is None
     assert attributes.message_deduplication_id is None
-    assert (
-        attributes.dead_letter_queue_source_arn
-        == raw_attributes["DeadLetterQueueSourceArn"]
-    )
+    assert attributes.dead_letter_queue_source_arn == raw_attributes["DeadLetterQueueSourceArn"]
 
 
 def test_decode_nested_s3_event():

tests/unit/parser/test_cloudformation_custom_resource.py

@@ -60,6 +60,14 @@ def test_cloudformation_custom_resource_update_event():
     assert model.old_resource_properties == raw_event["OldResourceProperties"]
 
 
+def test_cloudformation_custom_resource_update_event_physical_id_missing():
+    raw_event = load_event("cloudformationCustomResourceUpdate.json")
+    del raw_event["PhysicalResourceId"]
+
+    with pytest.raises(ValidationError):
+        CloudFormationCustomResourceUpdateModel(**raw_event)
+
+
 def test_cloudformation_custom_resource_update_event_invalid():
     raw_event = load_event("cloudformationCustomResourceUpdate.json")
     raw_event["OldResourceProperties"] = ["some_data"]
@@ -82,6 +90,14 @@ def test_cloudformation_custom_resource_delete_event():
     assert model.resource_properties == raw_event["ResourceProperties"]
 
 
+def test_cloudformation_custom_resource_delete_event_physical_id_missing():
+    raw_event = load_event("cloudformationCustomResourceUpdate.json")
+    del raw_event["PhysicalResourceId"]
+
+    with pytest.raises(ValidationError):
+        CloudFormationCustomResourceUpdateModel(**raw_event)
+
+
 def test_cloudformation_custom_resource_delete_event_invalid():
     raw_event = load_event("cloudformationCustomResourceDelete.json")
     raw_event["ResourceProperties"] = ["some_data"]

tests/unit/parser/test_sqs.py

@@ -117,7 +117,4 @@ def test_sqs_dlq_trigger_event():
     convert_time = int(round(attributes.SentTimestamp.timestamp() * 1000))
     assert convert_time == int(raw_record["attributes"]["SentTimestamp"])
 
-    assert (
-        attributes.DeadLetterQueueSourceArn
-        == raw_record["attributes"]["DeadLetterQueueSourceArn"]
-    )
+    assert attributes.DeadLetterQueueSourceArn == raw_record["attributes"]["DeadLetterQueueSourceArn"]