CloudFront object invalidation in Python

by Drew Engelson — 22 Oct 2010

Amazon Web Services enables the management of all AWS services through some pretty powerful REST APIs. CloudFront is Amazon's content delivery network offering which allows you to serve files out of S3 buckets with much better performance. S3 files are cached at edge servers so that users may download them more quickly. The service plays nice with cache-control and expiration headers, but what happens when you need to delete or update some cached files from the edge immediately?

CloudFront object invalidation is a recent addition... here's how to do it in Python.

AWS REST authentication

One of the first things you'll need to do for each request is generate an authentication signature based on your AWS credentials. So let's define a function for that.

import base64
import datetime
import hmac
import hashlib

def get_aws_auth():
    date = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
    auth = 'AWS %s:%s' % (
        AWS_ACCESS_KEY,
        base64.b64encode(
            hmac.new(
                AWS_SECRET_ACCESS_KEY.encode('utf-8'),
                date.encode('utf-8'),
                hashlib.sha1
            ).digest()
        )
    )
    return date, auth

print get_aws_auth()('Thu, 22 Oct 2010 06:08:48 GMT', 'AWS 0PN5J17HBGZHT7JJ3X82:4cP0hCJsdCxTJ1jPXo7+e/YSu0g=')

We can now use get_aws_auth() to sign the following REST requests. Next, we'll create an invalidation request.

Invalidating CloudFront objects

The invalidation handler expects a POST request with an xml payload containing a list of file paths to be expired.

import urllib2

host = 'cloudfront.amazonaws.com'
path = '/2010-08-01/distribution/%s/invalidation' % AWS_CF_DISTRIBUTION_ID

# create authentication signature
date, auth = get_aws_auth()

files = ['/file-01.mp4', '/file-02.mp4', '/file-03.mp4']

# create xml data to post
data = ['<?xml version="1.0" encoding="UTF-8"?>']
data.append('<InvalidationBatch>')
data.append('%s' % ''.join(['<Path>/%s</Path>' % f for f in files]))
data.append('<CallerReference>%s</CallerReference>' % date)
data.append('</InvalidationBatch>')
data = ''.join(data)

# define required http headers
headers = {
    'Host': host,
    'Authorization' : auth,
    'Date': date,
    'Content-Type': 'text/xml',
    'Content-Length': len(data),
}

# make REST request, capture 201 (success) response
request_url = 'https://%s%s' % (host, path)
req = urllib2.Request(request_url, data, headers=headers)
try:
    response = urllib2.urlopen(req)
except urllib2.HTTPError, e:
    print "HTTP Error 201: Created"

Of course, you probably also want to check the status of your invalidation request

Retrieving invalidation request status

Invalidation requests can take up to 15 minutes. Hit the GET invalidation API to check on the status of your recent requests.

import urllib2
from xml.dom import minidom

host = 'cloudfront.amazonaws.com'
path = '/2010-08-01/distribution/%s/invalidation?MaxItems=2' % AWS_CF_DISTRIBUTION_ID

# create authentication signature
date, auth = get_aws_auth()

# define required http headers
headers = {
    'Host': host,
    'Authorization': auth,
    'Date': date,
}

# send REST request and print results
request_url = 'https://%s%s' % (host, path)
req = urllib2.Request(request_url, headers=headers)
response = urllib2.urlopen(req)
xml = response.read()
print minidom.parseString(xml).toprettyxml(indent=' '*4)
<?xml version="1.0" ?>
<InvalidationList xmlns="http://cloudfront.amazonaws.com/doc/2010-08-01/">
    <Marker/>
    <NextMarker>
        I1Z3T4GFYRIAZU
    </NextMarker>
    <MaxItems>
        2
    </MaxItems>
    <IsTruncated>
        true
    </IsTruncated>
    <InvalidationSummary>
        <Id>
            I2HD8ADHMH7R
        </Id>
        <Status>
            InProgress
        </Status>
    </InvalidationSummary>
    <InvalidationSummary>
        <Id>
            IEFHBD78AAQN
        </Id>
        <Status>
            Completed
        </Status>
    </InvalidationSummary>
</InvalidationList>

The easy way: Use boto

As of version 2.0b3, the kick-ass boto library (Python interface to Amazon Web Services) has added support for invalidating CloudFront objects.

from boto.cloudfront import CloudFrontConnection

files = ['/file-01.mp4', '/file-02.mp4', '/file-03.mp4']

conn = CloudFrontConnection(AWS_ACCESS_KEY, AWS_SECRET_ACCESS_KEY)
print conn.create_invalidation_request(AWS_CF_DISTRIBUTION_ID, files)
<boto.cloudfront.invalidation.InvalidationBatch object at 0x1012b6390>

So short and sweet.