« Back to Index

[Fastly Purge with Python]

View original Gist on GitHub

Tags: #python #tornado #cli #purge #cache #fastly #cdn

Fastly Purge with Python.py

#!/usr/bin/env python

# standard library modules

import argparse
import re
import sys
from urllib.parse import urlparse

# external modules

from bf_rig import settings

# third party modules

from tornado.httpclient import AsyncHTTPClient
from tornado.ioloop import IOLoop

AsyncHTTPClient.configure(None, defaults=dict(user_agent='BuzzFeedSiteAdminUI'))
http_client = AsyncHTTPClient()

# mapping of subdomain to fastly key
envs = {'stage': 'www_stage',
        'www-stage': 'www_stage',
        'www': 'www_prod'}


def stage_creds():
    username = settings.get('stage_username')
    password = settings.get('stage_password')
    return f'{username}:{password}@'


def headers(subdomain):
    service = envs[subdomain]
    fastly_key = settings.get(f'fastly_api_{service}_token')

    return {'Fastly-Key': fastly_key,
            'Content-Type': 'application/x-www-form-urlencoded'}


async def purge_url(url_to_purge):
    '''Purge given URL for the buzzfeed.com host.'''

    print_url = url_to_purge  # we might mutate the original value with stage creds
    parsed_url = urlparse(url_to_purge)
    match = re.match('^([^.]+).buzzfeed.com', parsed_url.netloc, flags=re.IGNORECASE)
    subdomain = match.group(1)

    if not subdomain:
        subdomain = 'stage'

    if re.search('^(?:www-)?stage', subdomain):
        # reconstruct url to include stage credentials
        url_to_purge = f'{parsed_url.scheme}://{stage_creds()}{parsed_url.netloc}{parsed_url.path}'

    print(f'url to purge: {print_url}')

    response = await http_client.fetch(url_to_purge,
                                       method='PURGE',
                                       headers=headers(subdomain),
                                       allow_nonstandard_methods=True,
                                       raise_error=False)

    body_decoded = response.body.decode('utf-8')

    if response.code != 200:
        print(f'something went wrong: ({response.code}) {body_decoded}')
        return

    print(body_decoded)


parser = argparse.ArgumentParser(description='Purge CDN')
parser.add_argument('-u', '--url', help='buzzfeed.com url to be purged', type=str)
args = parser.parse_args()

if not args.url:
    print('-u, --url flag missing. see -h, --help for more information.')
    sys.exit(1)

io_loop = IOLoop.current()
io_loop.run_sync(lambda: purge_url(args.url))