Custom Resource Helper (crhelper)
crhelper simplifies authoring CloudFormation Custom Resources, implementing best practices for handling responses to CloudFormation, exception and timeout trapping, and detailed configurable logging. It is an open-source project actively maintained by AWS CloudFormation. The current version is 2.0.12, with regular updates to its PyPI package.
Warnings
- gotcha crhelper is explicitly not intended for use with AWS CDK's `Provider` construct. Using it this way can lead to unexpected behavior or conflicts.
- gotcha When testing locally (e.g., with `sam local`), `crhelper` attempts to send a response to CloudFormation, which fails if no actual stack or pre-signed URL exists. This can obscure actual business logic errors.
- gotcha If a `create` event handler fails (e.g., due to `helper.init_failure`), `crhelper` might not generate or propagate a `PhysicalResourceId` correctly, which can lead to CloudFormation rollback issues (e.g., attempting a `delete` on a non-existent resource).
- gotcha Enabling the polling feature for long-running custom resources requires additional IAM permissions for the Lambda function's execution role to manage Lambda permissions and EventBridge rules. Failing to provide these will cause polling to fail.
Install
-
pip install crhelper -
pip install -t . crhelper
Imports
- CfnResource
from crhelper import CfnResource
Quickstart
import logging
import os
from crhelper import CfnResource
logger = logging.getLogger(__name__)
# Initialise the helper, all inputs are optional
helper = CfnResource(json_logging=True, log_level='INFO', boto_level='CRITICAL')
# Example function to be called by the helper
@helper.create
@helper.update
def create_update_resource(event, context):
logger.info("Got Create/Update event")
# Access properties from the CloudFormation event
my_property = event['ResourceProperties'].get('MyProperty', 'default')
logger.info(f"MyProperty: {my_property}")
# Simulate creating/updating a resource
physical_resource_id = f"my-custom-resource-{my_property}"
# Optionally return an ID that will be used for the PhysicalResourceId
# if None is returned, an ID will be generated. The value is used in subsequent Update/Delete events.
helper.Data['Result'] = 'Success'
return physical_resource_id
@helper.delete
def delete_resource(event, context):
logger.info("Got Delete event")
# Access the PhysicalResourceId of the resource to be deleted
physical_resource_id = event.get('PhysicalResourceId')
logger.info(f"Deleting resource: {physical_resource_id}")
# Simulate deleting the resource
# No return value needed for delete
def handler(event, context):
logger.info("Lambda handler invoked")
helper(event, context)
# Example of how to run this locally for testing purposes (outside Lambda)
if __name__ == '__main__':
# Mock event and context for local testing
mock_event = {
'RequestType': 'Create',
'ResponseURL': 'http://example.com/',
'StackId': 'arn:aws:cloudformation:us-east-1:123456789012:stack/MyStack/UUID',
'RequestId': 'uniqueid123',
'LogicalResourceId': 'MyCustomResource',
'ResourceType': 'Custom::MyResource',
'ResourceProperties': {
'ServiceToken': 'arn:aws:lambda:us-east-1:123456789012:function:my-cr-lambda',
'MyProperty': 'test-value'
}
}
mock_context = type('Context', (object,), {
'function_name': 'test-func',
'invoked_function_arn': 'arn:aws:lambda:us-east-1:123456789012:function:test-func',
'aws_request_id': 'reqid123',
'log_group_name': '/aws/lambda/test-func',
'log_stream_name': '2026/04/14/[$LATEST]uuid',
'memory_limit_in_mb': '128',
'get_remaining_time_in_millis': lambda: 60000 # Mock 60 seconds remaining
})()
print("--- Simulating Create Event ---")
# In a real scenario, helper would send response to ResponseURL
# For local test, we just call the handler and check logs
handler(mock_event, mock_context)
print("\n--- Simulating Update Event ---")
mock_event['RequestType'] = 'Update'
mock_event['PhysicalResourceId'] = 'my-custom-resource-test-value' # Must be present for Update/Delete
handler(mock_event, mock_context)
print("\n--- Simulating Delete Event ---")
mock_event['RequestType'] = 'Delete'
handler(mock_event, mock_context)