Delete Unused AWS NAT Gateways

import boto3 from botocore.exceptions import ( BotoCoreError, ClientError, NoCredentialsError, PartialCredentialsError, EndpointConnectionError, ) creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def list_nat_gateways_for_region(ec2_client, region_name): nat_gateways_with_regions = [] try: response = ec2_client.describe_nat_gateways() if response and 'NatGateways' in response and len(response['NatGateways']) > 0: for nat_gateway in response['NatGateways']: nat_gateway_info = { "NatGatewayId": nat_gateway['NatGatewayId'], "Region": region_name, "State": nat_gateway['State'] } nat_gateways_with_regions.append(nat_gateway_info) #print(nat_gateway_info) else: #print(f"No NAT Gateways found in region {region_name}.") p=1 # Dummy line except (NoCredentialsError, PartialCredentialsError, EndpointConnectionError, ClientError, BotoCoreError, Exception) as e: print(f"Error in region {region_name}: {str(e)}") return nat_gateways_with_regions def display_nat_gateways(data): # Initialize table with the desired structure and headers table = context.newtable() table.title = "NAT Gateways Information" table.num_cols = 3 # Number of columns according to headers table.num_rows = 1 # Starts with one row for headers table.has_header_row = True # Define header names based on the new structure headers = ["NAT Gateway ID", "Region", "State"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Sort the NAT gateway data by region for better organization data.sort(key=lambda x: x["Region"]) # Populate the table with NAT gateway data for row_num, entry in enumerate(data, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each NAT gateway entry values = [entry["NatGatewayId"], entry["Region"], entry["State"]] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) #region = 'us-east-1' # You can set this to None to check all regions region = None # Hardcoded for one time result all_nat_gateways = [] if region: ec2_client = boto3.client('ec2',aws_access_key_id=access_key,aws_secret_access_key=secret_key, region_name=region) all_nat_gateways.extend(list_nat_gateways_for_region(ec2_client, region)) else: for region_name in regions: ec2_client = boto3.client('ec2',aws_access_key_id=access_key,aws_secret_access_key=secret_key, region_name=region_name) all_nat_gateways.extend(list_nat_gateways_for_region(ec2_client, region_name)) display_nat_gateways(all_nat_gateways)

import boto3 from datetime import datetime, timedelta from botocore.exceptions import ClientError, BotoCoreError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] unused_days = 7 # Hardcoded for One time Result def check_unused_nat_gateways_for_region(nat_gateways_list): unused_nat_gateways = [] # Check if the list is empty or not if not nat_gateways_list: print("No NAT gateways received for processing.") return unused_nat_gateways print(f"Received {len(nat_gateways_list)} NAT gateways for processing.") for nat_gateway_info in nat_gateways_list: region_name = nat_gateway_info['Region'] nat_gateway_id = nat_gateway_info['NatGatewayId'] ec2_client = boto3.client('ec2', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region_name) cloudwatch = boto3.client('cloudwatch', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region_name) try: response = cloudwatch.get_metric_data( MetricDataQueries=[ { 'Id': 'm1', 'MetricStat': { 'Metric': { 'Namespace': 'AWS/NATGateway', 'MetricName': 'BytesOutToDestination', 'Dimensions': [ { 'Name': 'NatGatewayId', 'Value': nat_gateway_info['NatGatewayId'] } ] }, 'Period': 86400 * unused_days, 'Stat': 'Sum' }, 'ReturnData': True } ], StartTime=datetime.now() - timedelta(days=unused_days), EndTime=datetime.now() ) if not response['MetricDataResults'][0]['Values']: unused_nat_gateways.append(nat_gateway_info) except (ClientError, BotoCoreError, Exception) as e: print(f"Error in region {region_name} for NAT Gateway {nat_gateway_id}: {str(e)}") # Print the total number of unused NAT gateways print(f"Out of {len(nat_gateways_list)} NAT gateways, {len(unused_nat_gateways)} are unused.") return unused_nat_gateways ''' all_nat_gateways = [ {'NatGatewayId': 'nat-0bc09626aff12105a', 'Region': 'us-east-1', 'State': 'pending'}, {'NatGatewayId': 'nat-0cee3df0c034c58f8', 'Region': 'us-east-1', 'State': 'deleted'}, {'NatGatewayId': 'nat-0b5177c47df82bc51', 'Region': 'us-east-1', 'State': 'deleted'} ] # passed down from previous task ''' unused_nat_gateways = check_unused_nat_gateways_for_region(all_nat_gateways) context.skip_sub_tasks=True