A simple AWS Lambda script which automatically invalidates URLS in a CloudFront distribution based on files uploaded to its associated S3 bucket. To use it, you need to create a DynamoDB table named {bucket}-change-tracker with two String columns named "request_path" and "eTag" to contain tracked changes.

cloudfront_auto_invalidator.py

import boto3
import json
import urllib
import time

cloudfront_client = boto3.client('cloudfront')

def get_change_tracker(bucket):
    tracker = None
    try:
        tracker = boto3.resource("dynamodb").Table(f"{bucket}-change-tracker")
    except Exception:
        pass
    return tracker

def get_cloudfront_distribution_id(bucket):
    
    bucket_origin = f"{bucket}.s3.amazonaws.com"
    cf_distro_id = None

    # Create a reusable Paginator
    paginator = cloudfront_client.get_paginator('list_distributions')

    # Create a PageIterator from the Paginator
    page_iterator = paginator.paginate()

    for page in page_iterator:
        for distribution in page['DistributionList']['Items']:
            for cf_origin in distribution['Origins']['Items']:
                    origin_name = cf_origin['DomainName']
                    print(f"Origin found {origin_name}")
                    if bucket_origin == origin_name:
                            cf_distro_id = distribution['Id']
                            print(f"The CF distribution ID for {bucket} is {cf_distro_id}")

    return cf_distro_id


# --------------- Main handler ------------------
def lambda_handler(event, context):
    '''
    Creates a cloudfront invalidation for content added to an S3 bucket
    '''
    # Log the the received event locally.
    # print("Received event: " + json.dumps(event, indent=2))

    # Get the object from the event.
    for record in event['Records']:
        # print(f"Processing record: {json.dumps(record, indent=2)}")
        bucket = record['s3']['bucket']['name']
        key = urllib.parse.unquote_plus(record['s3']['object']['key'])
        eTag = record["s3"]["object"]["eTag"]

        if not key.startswith('/'):
            key = '/' + key
        
        tracker = get_change_tracker(bucket)
    
        if tracker:
            item = None
            try:
                query_condition = boto3.dynamodb.conditions.Key("request_path").eq(key)
                items = tracker.query(KeyConditionExpression=query_condition)
            except Exception as e:
                print(f"Error retrieving change tracking for {key} from bucket {bucket}. Event {json.dumps(event, indent=2)}")
                raise
            
            do_invalidate = True
            if record["eventName"] == "ObjectRemoved:Delete":
                if len(items["Items"]) > 0:
                    print("Action is a DELETE of a tracked file")
                    for item in items["Items"]:
                        print(f"Removing tracking of item: {item['request_path']}: {item['eTag']}")
                        tracker.delete_item(Key=item)
                else:
                    print("Action is a DELETE of an untracked file")
            elif record["eventName"] == "ObjectCreated:Put":
                if len(items["Items"]) > 0:
                    if eTag == items["Items"][0]["eTag"]:
                        print("Action is a PUT with no changes")
                        do_invalidate = False
                    else:
                        print("Action is a PUT with file changes")
                        tracker.update_item(Key={
                            "request_path": key,
                            "eTag": eTag
                        })
                else:
                    print("Action is a PUT of a new file")
                    tracker.put_item(Item={
                        "request_path": key,
                        "eTag": eTag
                    })
                    
            cf_distro_id = get_cloudfront_distribution_id(bucket)
        
            if cf_distro_id:
                if do_invalidate:
                    print(f"Creating invalidation for {key} on Cloudfront distribution {cf_distro_id}")
                    invalidation_batch = {
                        'Paths': {
                            'Quantity': 1,
                            'Items': [key]
                            },
                        'CallerReference': str(time.time())
                    }

                    try:
                        invalidation = cloudfront_client.create_invalidation(
                            DistributionId=cf_distro_id,
                            InvalidationBatch=invalidation_batch)
                        id = invalidation['Invalidation']['Id']
                        status = invalidation['Invalidation']['Status']
                        print(f"Submitted invalidation. ID {id} Status {status}")
                    except Exception:
                        print(f"Error processing object {key} from bucket {bucket}. Event {json.dumps(event, indent=2)}")
                        raise
                else:
                    print(f"Not creating invalidation for object {key}")
            else:
                print(f"Bucket {bucket} does not appear to be an origin for a Cloudfront distribution")
        else:
            print(f"Bucket {bucket} does not appear to have an associated DynamoDB tracking table.")
            print(f"Create a DynamoDB table named {bucket}-change-tracker with the following keys:")
            print(f"  request_path: String")
            print(f"  eTag: String")

    return 'Success'