Optimizing AWS CloudFront Distributions with Compression

import boto3 creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Initialize the CloudFront client client = boto3.client('cloudfront',aws_access_key_id=access_key,aws_secret_access_key=secret_key) try: # Use a paginator for the list_distributions call paginator = client.get_paginator('list_distributions') all_distributions = [] # Iterate over each page of results for page in paginator.paginate(): # Check if 'Items' key exists in the current page distributions_on_page = page.get('DistributionList', {}).get('Items', []) # Extend the all_distributions list with IDs from the current page all_distributions.extend([dist['Id'] for dist in distributions_on_page]) # Check if there are any distributions and print appropriate messages if all_distributions: print(f"Found {len(all_distributions)} CloudFront distributions. Here are their IDs:") for dist_id in all_distributions: print(dist_id) else: print("No CloudFront distributions found.") except client.exceptions.NoSuchDistribution as e: print("Error: No such distribution found.") print(f"Details: {str(e)}") except client.exceptions.AccessDenied as e: print("Error: Access denied. Make sure you have the appropriate permissions.") print(f"Details: {str(e)}") except Exception as e: print("An unexpected error occurred while fetching the CloudFront distributions.") print(f"Details: {str(e)}")

import boto3 creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] ''' # List of all distributions received from the previous task all_distributions = ['E2W06YV1I68DDO', 'E16ONX1GUDSAFS'] ''' # Initialize the CloudFront client client = boto3.client('cloudfront',aws_access_key_id=access_key,aws_secret_access_key=secret_key) # List to store distributions that need compression enabled distributions_needing_compression = [] # Check if there are any distributions passed to the task if not all_distributions: print("No CloudFront distributions provided to the task.") else: try: # Loop through each distribution ID for dist_id in all_distributions: # Fetch the distribution's configuration dist_config_response = client.get_distribution_config(Id=dist_id) cache_behaviors = dist_config_response['DistributionConfig']['DefaultCacheBehavior'] # Check if compression is not enabled for the distribution if 'Compress' not in cache_behaviors or not cache_behaviors['Compress']: distributions_needing_compression.append(dist_id) # Print the results if distributions_needing_compression: print(f"{len(distributions_needing_compression)} distributions need compression enabled:") for dist_id in distributions_needing_compression: print(dist_id) else: print("All provided distributions already have compression enabled.") # Handle specific exceptions for service errors except client.exceptions.NoSuchDistribution as e: print("Error: One or more distributions not found.") print(f"Details: {str(e)}") except client.exceptions.AccessDenied as e: print("Error: Access denied. Ensure you have the appropriate permissions.") print(f"Details: {str(e)}") # General exception handling for unforeseen errors except Exception as e: print("An unexpected error occurred while processing the CloudFront distributions.") print(f"Details: {str(e)}") context.proceed = False

import boto3 creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] ''' # List of distributions needing compression received from the previous task distributions_needing_compression = ['E2W06YV1I68DDO', 'E16ONX1GUDSAFS'] ''' # Initialize the CloudFront client client = boto3.client('cloudfront',aws_access_key_id=access_key,aws_secret_access_key=secret_key) # Check if there are any distributions passed to the task if not distributions_needing_compression: print("No CloudFront distributions provided to the task for enabling compression.") else: try: # Counter to keep track of the number of distributions updated updated_count = 0 # Loop through each distribution ID for dist_id in distributions_needing_compression: # Fetch the distribution's configuration dist_config_response = client.get_distribution_config(Id=dist_id) dist_config = dist_config_response['DistributionConfig'] # Update the distribution to enable compression dist_config['DefaultCacheBehavior']['Compress'] = True client.update_distribution( Id=dist_id, IfMatch=dist_config_response['ETag'], DistributionConfig=dist_config ) print(f"Enabled compression for distribution {dist_id}") updated_count += 1 # Print a summary of the results print(f"\nSummary:\nUpdated {updated_count} out of {len(distributions_needing_compression)} distributions.") except client.exceptions.NoSuchDistribution as e: print("Error: One or more distributions not found.") print(f"Details: {str(e)}") except client.exceptions.AccessDenied as e: print("Error: Access denied. Ensure you have the appropriate permissions.") print(f"Details: {str(e)}") except client.exceptions.InvalidIfMatchVersion as e: print("Error: The If-Match version is invalid or mismatched.") print(f"Details: {str(e)}") except Exception as e: print("An unexpected error occurred while updating the CloudFront distributions.") print(f"Details: {str(e)}")