feat(event_source): add CloudFormationCustomResourceEvent data class.

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,45 @@
+from typing import Any, Dict, Literal
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class CloudFormationCustomResourceEvent(DictWrapper):
+    @property
+    def request_type(self) -> Literal["Create", "Update", "Delete"]:
+        return 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) -> str:
+        return self.get("PhysicalResourceId") or ""
+
+    @property
+    def resource_type(self) -> str:
+        return self["ResourceType"]
+
+    @property
+    def resource_properties(self) -> Dict[str, Any]:
+        return self.get("ResourceProperties") or {}
+
+    @property
+    def old_resource_properties(self) -> Dict[str, Any]:
+        return self.get("OldResourceProperties") or {}

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,14 @@ 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 hl_lines="11 13 15 17 19"
+    --8<-- "examples/event_sources/src/cloudformation_custom_resource_handler.py"
+    ```
+
 ### CloudWatch Dashboard Custom Widget
 
 === "app.py"

examples/event_sources/src/cloudformation_custom_resource_handler.py

@@ -0,0 +1,43 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import (
+    CloudFormationCustomResourceEvent,
+    event_source,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+logger = Logger()
+
+
+@event_source(data_class=CloudFormationCustomResourceEvent)
+def lambda_handler(event: CloudFormationCustomResourceEvent, context: LambdaContext):
+    request_type = event.request_type
+
+    if request_type == "Create":
+        return on_create(event)
+    if request_type == "Update":
+        return on_update(event)
+    if request_type == "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}.")
+    # ...

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,29 @@
+import pytest
+
+from aws_lambda_powertools.utilities.data_classes import (
+    CloudFormationCustomResourceEvent,
+)
+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 == raw_event["RequestType"]
+    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.response_url == raw_event["ResponseURL"]
+    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("cloudformationCustomResourceDelete.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"]