/
cloudfront_invalidation.py
executable file
·106 lines (81 loc) · 3.15 KB
/
cloudfront_invalidation.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
#!/usr/bin/python
import boto3
import botocore
import argparse
import time
import sys
from pprint import pprint
parser = argparse.ArgumentParser(description='Clear cloudfront CDN for a file')
parser.add_argument(
'distribution', help="Distribution to delete from", default="DEFAULT")
parser.add_argument('--path', help="path to expire", action="append")
parser.add_argument('--request', help="Request ID to track")
parser.add_argument('--profile', help="AWS Profile to use")
args = parser.parse_args()
if not (args.request or args.path):
parser.error('No action requested, add --request or at least one --path')
if args.profile:
print(("Setting session to {}".format(args.profile)))
boto3.setup_default_session(profile_name=args.profile)
cf = boto3.client('cloudfront')
distribution = args.distribution
matched = 0
def find_distro_id(cf, distribution):
distros = cf.list_distributions()
for distro in distros['DistributionList']['Items']:
if distro['Id'] == distribution:
print((" -> Distro found by ID {}".format(distro['Id'])))
return distro['Id']
if distro['DomainName'] == distribution:
print(
(" -> Distro found by Domain {} == {} ".format(distribution, distro['Id'])))
return distro['Id']
if distro['Aliases']['Quantity']:
for alias in distro['Aliases']["Items"]:
if alias == distribution:
print(
(" -> Distro found by Alias {} == {} ".format(distribution, distro['Id'])))
return distro['Id']
return False
distribution = find_distro_id(cf, distribution)
if not distribution:
print("Couldn't find matching disto")
sys.exit(5)
if args.path:
InvalidationBatch = {
'Paths': {
'Quantity': len(args.path),
'Items': args.path
},
'CallerReference': str(time.time()).replace(".", "")
}
try:
resp = cf.create_invalidation(
DistributionId=distribution, InvalidationBatch=InvalidationBatch)
except botocore.exceptions.ClientError as error:
print('Error creating invalidation:')
print(("AWS Error {}: {}".format(
error.response['Error']['Code'], error.response['Error']['Message'])))
sys.exit(5)
elif args.request:
try:
resp = cf.get_invalidation(
DistributionId=distribution, Id=args.request)
except botocore.exceptions.ClientError as error:
print('Error fetching response:')
print(("AWS Error {}: {}".format(
error.response['Error']['Code'], error.response['Error']['Message'])))
sys.exit(5)
else:
sys.exit(5)
print(('Invalidation {} now in progress. (you can ctrl-c out of this process and recheck with --request later) '.format(
resp['Invalidation']['Id'])))
while resp['Invalidation']['Status'] == "InProgress":
sys.stdout.write('.')
sys.stdout.flush()
time.sleep(5)
resp = cf.get_invalidation(
DistributionId=distribution, Id=resp['Invalidation']['Id'])
print("!")
print(('Invalidation {} {}: '.format(
resp['Invalidation']['Id'], resp['Invalidation']['Status'])))