AWS Boto3 is the Python SDK for AWS. Boto3 can be used to directly interact with AWS resources from Python scripts. In this tutorial, we will look at how we can use the Boto3 library to perform various operations on AWS CloudFront. CloudFront is a CDN (content delivery network) service built for high performance, security, and developer convenience. CloudFront can be used to deliver content to users at high speeds. It can also be used to protect your content from DDoS attacks.

Table of contents

Prerequisites

  • Python3
  • Boto3: Boto3 can be installed using pip: pip install boto3
  • AWS Credentials: If you haven’t setup AWS credentials before, this resource from AWS is helpful.

How to create a distribution?

To use CloudFront, we need to create a distribution. A distribution is a collection of settings that tell CloudFront how to handle requests for your content. A distribution can be created using the following code:


import boto3
from time import time, sleep


client = boto3.client("cloudfront")


def create_distribution():
    """Create a Cloudfront Distribution."""
    cloudfront_deployment_name = "learnaws-cf-test"
    s3_bucket_name = "learnaws-test-versioning-s3"

    cloudfront_dist_config = {
        "CallerReference": cloudfront_deployment_name,
        "Comment": "LearnAWS Cloudfront Distribution",
        "Origins": {
            "Quantity": 1,
            "Items": [{
                "Id": s3_bucket_name,
                "DomainName": s3_bucket_name + ".s3.amazonaws.com",
                "S3OriginConfig": {
                    "OriginAccessIdentity": ""
                }
            }]
        },
        "DefaultCacheBehavior": {
            "TargetOriginId": s3_bucket_name,
            "ViewerProtocolPolicy": "redirect-to-https",
            "TrustedSigners": {
                "Quantity": 0,
                "Enabled": False
            },
            "ForwardedValues": {
                "Cookies": {"Forward": "all"},
                "Headers": {"Quantity": 0},
                "QueryString": False,
                "QueryStringCacheKeys": {"Quantity": 0}
            },
            "DefaultTTL": 86400,
            "MinTTL": 3600
        },
        "Comment": "",
        "Enabled": True,
    }

    return client.create_distribution(DistributionConfig=cloudfront_dist_config)

In this particular example, we have created a CloudFront distribution for a S3 bucket as the origin. To learn more about the various parameters that can be used to create a distribution, refer to the AWS documentation.

How to list all distributions?

To list all distributions, we can use the following code:


def list():
    """Lists all distributions in Cloudfront.

    Relies on NextMarker to paginate through all distributions.

    Returns:
        List: List of distributions.
    """
    response = client.list_distributions()
    distributions = []
    while True:
        for dist in response["DistributionList"]["Items"]:
            distributions.append(dist)
        if "NextMarker" not in response["DistributionList"]:
            break

        response = client.list_distributions(Marker=response["DistributionList"]["NextMarker"])
    return distributions

The output of this function is:

list()

[{'Id': 'E1Z2Y3X4Z5', 'ARN': 'arn:aws:cloudfront::xxxxx:distribution/E1Z2Y3X4Z5', 'Status': 'InProgress', 'LastModifiedTime': datetime.datetime(2021, 5, 12, 3, 14, 32, tzinfo=tzutc()), 'InProgressInvalidationBatches': 0, 'DomainName': 'd1z2y3x4z5.cloudfront.net', 'ActiveTrustedSigners': {'Enabled': False, 'Quantity': 0}, 'DistributionConfig': {'CallerReference': 'learnaws-cf-test', 'Aliases': {'Quantity': 0}, 'DefaultRootObject': '', 'Origins': {'Quantity': 1, 'Items': [{'Id': 'learnaws-test-versioning-s3', 'DomainName': 'learnaws-test-versioning-s3.s3.amazonaws.com', 'S3OriginConfig': {'OriginAccessIdentity': ''}}]}, 'DefaultCacheBehavior': {'TargetOriginId': 'learnaws-test-versioning-s3', 'ForwardedValues':
....

}]

How to retrieve an existing distribution?

To retrieve an existing distribution, we need to provide the distribution ID. We can use the following code to retrieve a distribution:


def get_distribution(distribution_id):
    return client.get_distribution(Id=distribution_id)

The output of this function is:

get_distribution("E1Z2Y3X4Z5")

{'Distribution': {'Id': 'E1Z2Y3X4Z5', 'ARN': 'arn:aws:cloudfront::xxxxx:distribution/E1Z2Y3X4Z5', 'Status': 'InProgress', 'LastModifiedTime': datetime.datetime(2021, 5, 12, 3, 14, 32, tzinfo=tzutc()), 'InProgressInvalidationBatches': 0, 'DomainName': 'd1z2y3x4z5.cloudfront.net', 'ActiveTrustedSigners': {'Enabled': False, 'Quantity': 0}, 'DistributionConfig': {'CallerReference': 'learnaws-cf-test', 'Aliases': {'Quantity': 0}, 'DefaultRootObject': '', 'Origins': {'Quantity': 1, 'Items': [{'Id': 'learnaws-test-versioning-s3', 'DomainName': 'learnaws-test-versioning-s3.s3.amazonaws.com', 'S3OriginConfig': {'OriginAccessIdentity': ''}}]}, 'DefaultCacheBehavior': {'TargetOriginId': 'learnaws-test-versioning-s3', 'ForwardedValues':
.....

How to update a distribution?

To update a distribution, we need to provide the distribution ID and the updated distribution configuration. The distribution ID can be obtained from the response of the create_distribution function. The updated distribution configuration can be obtained from the response of the get_distribution function. Before we can update a distribution, we need to provide the ETag of the existing distribution configuration.

We can use the following code to update a distribution:


def update():
    distribution_config = client.get_distribution_config(Id=distribution_id)

    # make the desired changes to the distribution configuration
    new_distribution_config = distribution_config["DistributionConfig"]
    new_distribution_config["Comment"] = "Updated Comment"

    # update the distribution with the new configuration
    cloudfront.update_distribution(
        DistributionConfig=new_distribution_config,
        Id=distribution_id,
        IfMatch=distribution_config["ETag"]
    )

How to create an invalidation?

To create an invalidation, we need to provide the distribution ID and the paths that need to be invalidated. The paths that need to be invalidated can be a list of paths or a wildcard. We can use the following code to create an invalidation:


import uuid


def invalidate(distribution_id, paths):
    """Create an invalidation for the given paths.

    Args:
        distribution_id (string): Distribution ID of the CloudFront distribution.
        paths (List): Array of paths to be invalidated.

    Returns:
        dict: Details of the invalidation object
    """
    return client.create_invalidation(
        DistributionId=distribution_id,
        InvalidationBatch={
            "Paths": {
                "Quantity": len(paths),
                "Items": paths
            },
            "CallerReference": str(uuid.uuid4())
        }
    )

The output of this function is:


{'ResponseMetadata': {'RequestId': '18761caa-3e2c-456a-be4a-50c64846d55b',
  'HTTPStatusCode': 201,
  'HTTPHeaders': {'x-amzn-requestid': '18761caa-3e2c-456a-be4a-50c64846d55b',
   'location': 'https://cloudfront.amazonaws.com/2020-05-31/distribution/E27D2Q31KB75SU/invalidation/I3UXUSHWOQO7I5',
   'content-type': 'text/xml',
   'content-length': '362',
   'date': 'Wed, 01 Feb 2023 04:34:38 GMT'},
  'RetryAttempts': 0},
 'Location': 'https://cloudfront.amazonaws.com/2020-05-31/distribution/E27D2Q31KB75SU/invalidation/I3UXUSHWOQO7I5',
 'Invalidation': {'Id': 'I3UXUSHWOQO7I5',
  'Status': 'InProgress',
  'CreateTime': datetime.datetime(2023, 2, 1, 4, 34, 39, 358000, tzinfo=tzutc()),
  'InvalidationBatch': {'Paths': {'Quantity': 1, 'Items': ['/*']},
   'CallerReference': '167522607898338'}}}

How to check if invalidation has completed?

To check if an invalidation has completed, we need to provide the distribution ID and the invalidation ID. We can use the following code to check if an invalidation has completed:


from time import sleep

def wait_invalidation_complete(distribution_id, invalidation_id):
    """Check if the invalidation has completed.

    Args:
        distribution_id (string): Distribution ID of the CloudFront distribution.
        invalidation_id (string): INvalidation ID of the invalidation.
    """

    while True:
        invalidation_status = client.get_invalidation(
            DistributionId=distribution_id,
            Id=invalidation_id
        )["Invalidation"]["Status"]

        if invalidation_status == "Completed":
            print("invalidation completed")
            return

        print("invalidation not yet completed, sleeping for 30 seconds")
        sleep(30)

Conclusion

In this tutorial, we learned how to create a CloudFront distribution, retrieve an existing distribution, update a distribution, create an invalidation, and check if an invalidation has completed.