This should be fun…
This is my python and boto3 hello world. I have never touched these let’s see what we get. Personally I wish perl had won.
With the said here is it the link I am going to use.
https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html
Installation
PythonBoto Mac:
sudo apt update
sudo apt install python3-pip
pip3 --version
pip3 install boto3
pip3 install --upgrade awscli
import ngw_change as ngw
#used for testing until it runs
# current_subnet = 'subnet-0f6cb2238d29ec05e'
# ngw.create_gateway.create_nat(current_subnet)
#describe_nat_gateways(tag, tag_values, max_items)
foo = ngw.describe_gateway.describe_nat_gateways('Name',['k8s-env-delete-me','k8s-env-delete-me2'], 10)
for bar in foo:
print (str(bar['SubnetId']) + "n")
# Use the job to set the key name RESOURCE_ENV for each resource. Create a dictionary
# entry for each resource key name listing the AWS "Name" of each affected natgateway
# RESOURCE_ENV='resource_env1'
# resource_dict = {
# 'resource_env1': ('k8s-env-delete-me') #, 'k8s-env1'),
# #'resource_env2': ('k8s-env0', 'k8s-env1')
# }
# with open("outfile", "w") as outfile:
# outfile.write("n".join(str(item) for item in foo))
#python3 -m ensurepip
import logging
import boto3
from datetime import date, datetime
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-west-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)
def json_datetime_serializer(obj):
"""
Helper method to serialize datetime fields
"""
if isinstance(obj, (datetime, date)):
return obj.isoformat()
raise TypeError("Type %s not serializable" % type(obj))
def describe_nat_gateways(tag, tag_values, max_items):
"""
Describes one or more of your NAT gateways.
"""
try:
# creating paginator object for describe_nat_gateways() method
paginator = vpc_client.get_paginator('describe_nat_gateways')
# creating a PageIterator from the paginator
response_iterator = paginator.paginate(
Filters=[{
'Name': f"tag:{tag}",
'Values': tag_values
}],
PaginationConfig={'MaxItems': max_items})
full_result = response_iterator.build_full_result()
nat_gateways_list = []
for page in full_result['NatGateways']:
nat_gateways_list.append(page)
except ClientError:
logger.exception('Could not describe NAT Gateways.')
raise
else:
return nat_gateways_list
if __name__ == '__main__':
print ("I am running as main")
# Constants
TAG = 'Name'
TAG_VALUES = ['k8s-env-delete-me']
MAX_ITEMS = 10
nat_gateways = describe_nat_gateways(TAG, TAG_VALUES, MAX_ITEMS)
logger.info('NAT Gateways Details: ')
for nat_gateway in nat_gateways:
logger.info(
json.dumps(nat_gateway, indent=4, default=json_datetime_serializer)
+ 'n')
print ('i am not running as main')
print("Value in built variable name is: ",__name__)
import logging
import boto3
from datetime import date, datetime
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-west-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)
def json_datetime_serializer(obj):
"""
Helper method to serialize datetime fields
"""
if isinstance(obj, (datetime, date)):
return obj.isoformat()
raise TypeError("Type %s not serializable" % type(obj))
def wait_nat_creation(nat_gateway_id):
"""
Check if successful state is reached every 15 seconds until a successful state is reached.
An error is returned after 40 failed checks.
"""
try:
waiter = vpc_client.get_waiter('nat_gateway_available')
waiter.wait(NatGatewayIds=[nat_gateway_id])
except ClientError:
logger.exception(f'Could not create the NAT gateway.')
raise
# def allocate_address():
# """
# Allocates an Elastic IP address to use with an NAT Gateway in a VPC.
# """
# try:
# response = vpc_client.allocate_address(Domain='vpc')
# except ClientError:
# logger.exception(f'Could not create the NAT gateway.')
# raise
# else:
# return response['AllocationId']
def create_nat(subnet_id):
"""
Creates a NAT gateway in the specified subnet.
"""
try:
# allocate IPV4 address for NAT gateway
public_ip_allocation_id = "eipalloc-05c7766e0e9d3c02c"
# create NAT gateway
response = vpc_client.create_nat_gateway(
#AllocationId=public_ip_allocation_id,
SubnetId=subnet_id,
TagSpecifications=[{
'ResourceType':
'natgateway',
'Tags': [{
'Key': 'Name',
'Value': 'k8s-env-delete-me2'
}]
}])
nat_gateway_id = response['NatGateway']['NatGatewayId']
# wait until the NAT gateway is available
wait_nat_creation(nat_gateway_id)
except ClientError:
logger.exception(f'Could not create the NAT gateway.')
raise
else:
return response
if __name__ == '__main__':
print ("I am running as main")
# Constants
SUBNET_ID = ''
logger.info(f'Creating a NAT gateway...')
nat = create_nat(SUBNET_ID)
logger.info(
f'NAT gateway created with: {json.dumps(nat, indent=4, default=json_datetime_serializer)}'
)
print ('i am not running as main')
print("Value in built variable name is: ",__name__)
from . import create_gateway
from . import describe_gateway
import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-2'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
format='%(asctime)s: %(levelname)s: %(message)s')
vpc_client = boto3.client("ec2", region_name=AWS_REGION)
def delete_nat_gateway(nat_gateway_id):
"""
Deletes the specified NAT gateway.
"""
try:
response = vpc_client.delete_nat_gateway(NatGatewayId=nat_gateway_id)
except ClientError:
logger.exception('Could not delete the NAT Gateway.')
raise
else:
return response
if __name__ == '__main__':
# Constants
NAT_GATEWAY_ID = 'nat-0dce41a56b339d240'
nat_gateway = delete_nat_gateway(NAT_GATEWAY_ID)
logger.info(f'NAT Gateway {NAT_GATEWAY_ID} is deleted successfully.')