AWS Lambda Cloudfront Invalidator.py

#!/usr/bin/env python
"""This script will respond to changes on S3 stored files by calling invalidating those files from their respective
Cloudfront distribution.

Currently suppports the following folders: `/artefacts/", `/images/` and `payloads--charges-utiles/`.
"""
__author__ = "Rémy Vanherweghem"
__email__ = "remy.vanherweghem@parl.gc.ca"
__version__ = "1.0.0"
__copyright__ = "Copyright 2021, The Parliamentary Budget Officer"
__status__ = "Production"

import boto3
import time
import os
import itertools

DISTRIBUTIONS={"images": "E1I6BNJHL5TZJN", "artefacts":"E1H2OO9CLJI98M"}

def invalidate_batch(distribution_id, path_list):
    client = boto3.client('cloudfront')
    invalidation = client.create_invalidation(DistributionId=distribution_id,
        InvalidationBatch={
            'Paths': {
                'Quantity': 1,
                'Items': path_list
        },
        'CallerReference': str(time.time())
    })


def lambda_handler(event, context):
    invalidations = []
    for request_items in event["Records"]:
        key = request_items["s3"]["object"]["key"]
        dist = DISTRIBUTIONS[key.split('/')[0]]
        if dist:
            invalidations.append({"path": "/" + key.split("/")[-1], "distribution": dist})

    group_func = lambda x: x["distribution"]
    map_func = lambda x: x["path"]
    for key, group in itertools.groupby(invalidations, group_func):
        invalidate_batch(key, list(map(map_func, list(group))))