PURPOSE

Sharing a collection of snippets to perform common AWS operations such as Upload and Cache-Invalidation for S3 buckets, using Python.

Prerequisites

You should have the following keys before starting:

• AWS Access Key
• AWS Secret Key
• AWS Distribution ID (optional, required for invalidating cache)

---

In a hurry? Grab this gist.

---

How to Upload a file with public access on an S3 Bucket

import traceback import boto3 from botocore.exceptions import NoCredentialsError class AWSActions: def __init__(self): self.ACCESS_KEY = '--your key--' self.SECRET_KEY = '--your secret key--' self.s3 = boto3.client('s3', aws_access_key_id=self.ACCESS_KEY, aws_secret_access_key=self.SECRET_KEY) def upload_to_aws(self, local_file, bucket, s3_file): """Uploads file to S3 Bucket with Public Read access Files uploaded have GET request have support Assumes that the uploaded file will be JSON Parameters ---------- local_file: str file (path) to be uploaded bucket: str aws s3 bucket s3_file: str path where file be placed in the s3 bucket Returns ------- bool True if file upload was successful """ try: s3_file = s3_file.strip('/') self.s3.upload_file(local_file, bucket, s3_file, ExtraArgs={ 'CacheControl' : 'max-age=1209600', 'ContentType': 'application/json', 'ACL': 'public-read'}) return True except FileNotFoundError: print("The file was not found") return False except NoCredentialsError: print("Credentials not available") return False except: print(traceback.format_exc()) return None if __name__ == '__main__': upload = AWSActions() uploaded = upload.upload_to_aws( local_file=r'pop.json', bucket='--your bucket name--', s3_file='/folder_path/file_name.json' ) print('Uploaded:', uploaded)

Note: If you don't want the access to be public, remove the ACL key in the ExtraArgs dictionary passed to upload_file function.

---

How to Invalidate Cache of a File in S3 Bucket

import time import boto3 class AWSActions: def __init__(self): self.ACCESS_KEY = '--your key--' self.SECRET_KEY = '--your secret key--' self.distributionId = '--your distribution id---' self.cf = boto3.client('cloudfront', aws_access_key_id=self.ACCESS_KEY, aws_secret_access_key=self.SECRET_KEY) def createInvalidation(self, paths): """Invalidates existing files on cdn This removes file from S3 cache, so that latest version of the file is served on request Note that it is more of a request than a command (s3 may or may not immediately update the file) If file changes are not reflected after invalidation request, pass in additional query parameter Example: https://cdn.yourcompany.com/folder_name/file_name.json - old file https://cdn.yourcompany.com/folder_name/file_name.json?rand_param=123junk - new file Parameters ---------- paths file paths on s3 bucket that need to be invalidated Returns ------- dict Invalidation request object """ try: milliseconds=str(round(time.time() * 1000)) print("CallerReference:", milliseconds) InvalidationBatch = { 'Paths': { 'Quantity': int(len(paths)), 'Items': paths }, 'CallerReference': milliseconds } invalidation = self.cf.create_invalidation( DistributionId=self.distributionId, InvalidationBatch=InvalidationBatch ) return invalidation except: return None if __name__ == '__main__': upload = AWSActions() invalidated = upload.createInvalidation( paths=[f'/folder_path/file_name.json'] ) print('Invalidated:', invalidated)