aws-cdk-custom-resources

import os from aws_cdk import ( Stack, App, Duration, ) from aws_cdk.aws_s3 import Bucket from aws_cdk.custom_resources import ( AwsCustomResource, AwsCustomResourcePolicy, PhysicalResourceId, ) from constructs import Construct class MyCustomResourceStack(Stack): def __init__(self, scope: Construct, id: str, **kwargs) -> None: super().__init__(scope, id, **kwargs) # Create an S3 bucket to interact with bucket = Bucket(self, "MyCustomResourceBucket") # Use AwsCustomResource to call S3 listObjectsV2 API # This construct makes direct SDK calls from CloudFormation s3_list_caller = AwsCustomResource( self, "S3ListCaller", on_create={ "service": "S3", "action": "listObjectsV2", "parameters": { "Bucket": bucket.bucket_name, }, "physical_resource_id": PhysicalResourceId.of(f"my-s3-lister-{bucket.bucket_name}"), }, on_update={ "service": "S3", "action": "listObjectsV2", "parameters": { "Bucket": bucket.bucket_name, }, "physical_resource_id": PhysicalResourceId.of(f"my-s3-lister-{bucket.bucket_name}"), }, # on_delete is optional; here it's not needed for a read-only action # For resources that create external entities, on_delete is critical for cleanup. policy=AwsCustomResourcePolicy.from_sdk_calls( resources=[bucket.bucket_arn, bucket.bucket_arn + "/*"] ), timeout=Duration.minutes(2) ) # You can retrieve outputs from the SDK call result # For listObjectsV2, it returns a list of contents. Here, we just get the Request ID. request_id = s3_list_caller.get_response_field("ResponseMetadata.RequestId") # In a real application, you might use this output, e.g., CfnOutput(self, "RequestId", value=request_id) app = App() MyCustomResourceStack(app, "MyCustomResourceExampleStack") app.synth()