2022-10-11 15:26:03 -07:00
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
import logging
2024-09-05 12:39:58 -04:00
from typing import Any , Dict , List , Optional
2022-10-11 15:26:03 -07:00
import boto3
from botocore . exceptions import ClientError
logger = logging . getLogger ( __name__ )
# snippet-start:[python.example_code.ec2.InstanceWrapper.class]
# snippet-start:[python.example_code.ec2.InstanceWrapper.decl]
2024-09-05 12:39:58 -04:00
class EC2InstanceWrapper :
""" Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface. """
2023-10-18 10:35:05 -07:00
2024-09-05 12:39:58 -04:00
def __init__ (
self , ec2_client : Any , instances : Optional [ List [ Dict [ str , Any ] ] ] = None
) - > None :
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
Initializes the EC2InstanceWrapper with an EC2 client and optional instances.
:param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level
access to AWS EC2 services.
:param instances: A list of dictionaries representing Boto3 Instance objects. These are high-level objects that
wrap instance actions.
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
self . ec2_client = ec2_client
self . instances = instances or [ ]
2022-10-11 15:26:03 -07:00
@classmethod
2024-09-05 12:39:58 -04:00
def from_client ( cls ) - > " EC2InstanceWrapper " :
"""
Creates an EC2InstanceWrapper instance with a default EC2 client.
:return: An instance of EC2InstanceWrapper initialized with the default EC2 client.
"""
ec2_client = boto3 . client ( " ec2 " )
return cls ( ec2_client )
2023-10-18 10:35:05 -07:00
# snippet-end:[python.example_code.ec2.InstanceWrapper.decl]
2022-10-11 15:26:03 -07:00
# snippet-start:[python.example_code.ec2.RunInstances]
2024-09-05 12:39:58 -04:00
def create (
self ,
image_id : str ,
instance_type : str ,
key_pair_name : str ,
security_group_ids : Optional [ List [ str ] ] = None ,
) - > List [ Dict [ str , Any ] ] :
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
Creates a new EC2 instance in the default VPC of the current account.
2022-10-11 15:26:03 -07:00
2024-09-05 12:39:58 -04:00
The instance starts immediately after it is created.
2022-10-11 15:26:03 -07:00
2024-09-05 12:39:58 -04:00
:param image_id: The ID of the Amazon Machine Image (AMI) to use for the instance.
2022-10-11 15:26:03 -07:00
:param instance_type: The type of instance to create, such as ' t2.micro ' .
2024-09-05 12:39:58 -04:00
:param key_pair_name: The name of the key pair to use for SSH access.
:param security_group_ids: A list of security group IDs to associate with the instance.
If not specified, the default security group of the VPC is used.
:return: A list of dictionaries representing Boto3 Instance objects representing the newly created instances.
2022-10-11 15:26:03 -07:00
"""
try :
instance_params = {
2024-09-05 12:39:58 -04:00
" ImageId " : image_id ,
2023-10-18 10:35:05 -07:00
" InstanceType " : instance_type ,
2024-09-05 12:39:58 -04:00
" KeyName " : key_pair_name ,
2022-10-11 15:26:03 -07:00
}
2024-09-05 12:39:58 -04:00
if security_group_ids is not None :
instance_params [ " SecurityGroupIds " ] = security_group_ids
response = self . ec2_client . run_instances (
2023-10-18 10:35:05 -07:00
* * instance_params , MinCount = 1 , MaxCount = 1
2024-09-05 12:39:58 -04:00
)
instance = response [ " Instances " ] [ 0 ]
self . instances . append ( instance )
waiter = self . ec2_client . get_waiter ( " instance_running " )
waiter . wait ( InstanceIds = [ instance [ " InstanceId " ] ] )
2022-10-11 15:26:03 -07:00
except ClientError as err :
2024-09-05 12:39:58 -04:00
params_str = " \n \t " . join (
f " { key } : { value } " for key , value in instance_params . items ( )
)
logger . error (
f " Failed to complete instance creation request. \n Request details: { params_str } "
2023-10-18 10:35:05 -07:00
)
2024-09-05 12:39:58 -04:00
error_code = err . response [ " Error " ] [ " Code " ]
if error_code == " InstanceLimitExceeded " :
logger . error (
(
f " Insufficient capacity for instance type ' { instance_type } ' . "
" Terminate unused instances or contact AWS Support for a limit increase. "
)
)
if error_code == " InsufficientInstanceCapacity " :
logger . error (
(
f " Insufficient capacity for instance type ' { instance_type } ' . "
" Select a different instance type or launch in a different availability zone. "
)
)
2022-10-11 15:26:03 -07:00
raise
2024-09-05 12:39:58 -04:00
return self . instances
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.RunInstances]
# snippet-start:[python.example_code.ec2.DescribeInstances]
2024-09-05 12:39:58 -04:00
def display ( self , state_filter : Optional [ str ] = " running " ) - > None :
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
Displays information about instances, filtering by the specified state.
2022-10-11 15:26:03 -07:00
2024-09-05 12:39:58 -04:00
:param state_filter: The instance state to include in the output. Only instances in this state
will be displayed. Default is ' running ' . Example states: ' running ' , ' stopped ' .
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
if not self . instances :
logger . info ( " No instances to display. " )
2022-10-11 15:26:03 -07:00
return
2024-09-05 12:39:58 -04:00
instance_ids = [ instance [ " InstanceId " ] for instance in self . instances ]
paginator = self . ec2_client . get_paginator ( " describe_instances " )
page_iterator = paginator . paginate ( InstanceIds = instance_ids )
2022-10-11 15:26:03 -07:00
try :
2024-09-05 12:39:58 -04:00
for page in page_iterator :
for reservation in page [ " Reservations " ] :
for instance in reservation [ " Instances " ] :
instance_state = instance [ " State " ] [ " Name " ]
# Apply the state filter (default is 'running')
if state_filter and instance_state != state_filter :
continue # Skip this instance if it doesn't match the filter
# Create a formatted string with instance details
instance_info = (
f " • ID: { instance [ ' InstanceId ' ] } \n "
f " • Image ID: { instance [ ' ImageId ' ] } \n "
f " • Instance type: { instance [ ' InstanceType ' ] } \n "
f " • Key name: { instance [ ' KeyName ' ] } \n "
f " • VPC ID: { instance [ ' VpcId ' ] } \n "
f " • Public IP: { instance . get ( ' PublicIpAddress ' , ' N/A ' ) } \n "
f " • State: { instance_state } "
)
print ( instance_info )
2022-10-11 15:26:03 -07:00
except ClientError as err :
logger . error (
2024-09-05 12:39:58 -04:00
f " Failed to display instance(s). : { ' ' . join ( map ( str , instance_ids ) ) } "
2023-10-18 10:35:05 -07:00
)
2024-09-05 12:39:58 -04:00
error_code = err . response [ " Error " ] [ " Code " ]
if error_code == " InvalidInstanceID.NotFound " :
logger . error (
" One or more instance IDs do not exist. "
" Please verify the instance IDs and try again. "
)
raise
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.DescribeInstances]
# snippet-start:[python.example_code.ec2.TerminateInstances]
2024-09-05 12:39:58 -04:00
def terminate ( self ) - > None :
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
Terminates instances and waits for them to reach the terminated state.
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
if not self . instances :
logger . info ( " No instances to terminate. " )
2022-10-11 15:26:03 -07:00
return
2024-09-05 12:39:58 -04:00
instance_ids = [ instance [ " InstanceId " ] for instance in self . instances ]
2022-10-11 15:26:03 -07:00
try :
2024-09-05 12:39:58 -04:00
self . ec2_client . terminate_instances ( InstanceIds = instance_ids )
waiter = self . ec2_client . get_waiter ( " instance_terminated " )
waiter . wait ( InstanceIds = instance_ids )
self . instances . clear ( )
for instance_id in instance_ids :
print ( f " • Instance ID: { instance_id } \n " f " • Action: Terminated " )
2022-10-11 15:26:03 -07:00
except ClientError as err :
2024-09-05 12:39:58 -04:00
logger . error (
f " Failed instance termination details: \n \t { str ( self . instances ) } "
2023-10-18 10:35:05 -07:00
)
2024-09-05 12:39:58 -04:00
error_code = err . response [ " Error " ] [ " Code " ]
if error_code == " InvalidInstanceID.NotFound " :
logger . error (
" One or more instance IDs do not exist. "
" Please verify the instance IDs and try again. "
)
2022-10-11 15:26:03 -07:00
raise
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.TerminateInstances]
# snippet-start:[python.example_code.ec2.StartInstances]
2024-09-05 12:39:58 -04:00
def start ( self ) - > Optional [ Dict [ str , Any ] ] :
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
Starts instances and waits for them to be in a running state.
2022-10-11 15:26:03 -07:00
:return: The response to the start request.
"""
2024-09-05 12:39:58 -04:00
if not self . instances :
logger . info ( " No instances to start. " )
return None
2022-10-11 15:26:03 -07:00
2024-09-05 12:39:58 -04:00
instance_ids = [ instance [ " InstanceId " ] for instance in self . instances ]
2022-10-11 15:26:03 -07:00
try :
2024-09-05 12:39:58 -04:00
start_response = self . ec2_client . start_instances ( InstanceIds = instance_ids )
waiter = self . ec2_client . get_waiter ( " instance_running " )
waiter . wait ( InstanceIds = instance_ids )
return start_response
2022-10-11 15:26:03 -07:00
except ClientError as err :
logger . error (
2024-09-05 12:39:58 -04:00
f " Failed to start instance(s): { ' , ' . join ( map ( str , instance_ids ) ) } "
2023-10-18 10:35:05 -07:00
)
2024-09-05 12:39:58 -04:00
error_code = err . response [ " Error " ] [ " Code " ]
if error_code == " IncorrectInstanceState " :
logger . error (
" Couldn ' t start instance(s) because they are in an incorrect state. "
" Ensure the instances are in a stopped state before starting them. "
)
2022-10-11 15:26:03 -07:00
raise
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.StartInstances]
# snippet-start:[python.example_code.ec2.StopInstances]
2024-09-05 12:39:58 -04:00
def stop ( self ) - > Optional [ Dict [ str , Any ] ] :
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
Stops instances and waits for them to be in a stopped state.
2022-10-11 15:26:03 -07:00
2024-09-05 12:39:58 -04:00
:return: The response to the stop request, or None if there are no instances to stop.
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
if not self . instances :
logger . info ( " No instances to stop. " )
return None
2022-10-11 15:26:03 -07:00
2024-09-05 12:39:58 -04:00
instance_ids = [ instance [ " InstanceId " ] for instance in self . instances ]
2022-10-11 15:26:03 -07:00
try :
2024-09-05 12:39:58 -04:00
# Attempt to stop the instances
stop_response = self . ec2_client . stop_instances ( InstanceIds = instance_ids )
waiter = self . ec2_client . get_waiter ( " instance_stopped " )
waiter . wait ( InstanceIds = instance_ids )
2022-10-11 15:26:03 -07:00
except ClientError as err :
logger . error (
2024-09-05 12:39:58 -04:00
f " Failed to stop instance(s): { ' , ' . join ( map ( str , instance_ids ) ) } "
2023-10-18 10:35:05 -07:00
)
2024-09-05 12:39:58 -04:00
error_code = err . response [ " Error " ] [ " Code " ]
if error_code == " IncorrectInstanceState " :
logger . error (
" Couldn ' t stop instance(s) because they are in an incorrect state. "
" Ensure the instances are in a running state before stopping them. "
)
2022-10-11 15:26:03 -07:00
raise
2024-09-05 12:39:58 -04:00
return stop_response
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.StopInstances]
# snippet-start:[python.example_code.ec2.DescribeImages]
2024-09-05 12:39:58 -04:00
def get_images ( self , image_ids : List [ str ] ) - > List [ Dict [ str , Any ] ] :
2022-10-11 15:26:03 -07:00
"""
Gets information about Amazon Machine Images (AMIs) from a list of AMI IDs.
2024-09-05 12:39:58 -04:00
:param image_ids: The list of AMI IDs to look up.
:return: A list of dictionaries representing the requested AMIs.
2022-10-11 15:26:03 -07:00
"""
try :
2024-09-05 12:39:58 -04:00
response = self . ec2_client . describe_images ( ImageIds = image_ids )
images = response [ " Images " ]
2022-10-11 15:26:03 -07:00
except ClientError as err :
2024-09-05 12:39:58 -04:00
logger . error ( f " Failed to stop AMI(s): { ' , ' . join ( map ( str , image_ids ) ) } " )
error_code = err . response [ " Error " ] [ " Code " ]
if error_code == " InvalidAMIID.NotFound " :
logger . error ( " One or more of the AMI IDs does not exist. " )
2022-10-11 15:26:03 -07:00
raise
2024-09-05 12:39:58 -04:00
return images
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.DescribeImages]
# snippet-start:[python.example_code.ec2.DescribeInstanceTypes]
2024-09-05 12:39:58 -04:00
def get_instance_types (
self , architecture : str = " x86_64 " , sizes : List [ str ] = [ " *.micro " , " *.small " ]
) - > List [ Dict [ str , Any ] ] :
2022-10-11 15:26:03 -07:00
"""
2024-09-05 12:39:58 -04:00
Gets instance types that support the specified architecture and size.
See https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeInstanceTypes.html
for a list of allowable parameters.
:param architecture: The architecture supported by instance types. Default: ' x86_64 ' .
:param sizes: The size of instance types. Default: ' *.micro ' , ' *.small ' ,
:return: A list of dictionaries representing instance types that support the specified architecture and size.
2022-10-11 15:26:03 -07:00
"""
try :
inst_types = [ ]
2024-09-05 12:39:58 -04:00
paginator = self . ec2_client . get_paginator ( " describe_instance_types " )
for page in paginator . paginate (
2023-10-18 10:35:05 -07:00
Filters = [
{
" Name " : " processor-info.supported-architecture " ,
" Values " : [ architecture ] ,
} ,
2024-09-05 12:39:58 -04:00
{ " Name " : " instance-type " , " Values " : sizes } ,
2023-10-18 10:35:05 -07:00
]
) :
inst_types + = page [ " InstanceTypes " ]
2022-10-11 15:26:03 -07:00
except ClientError as err :
logger . error (
2024-09-05 12:39:58 -04:00
f " Failed to get instance types: { architecture } , { ' , ' . join ( map ( str , sizes ) ) } "
2023-10-18 10:35:05 -07:00
)
2024-09-05 12:39:58 -04:00
error_code = err . response [ " Error " ] [ " Code " ]
if error_code == " InvalidParameterValue " :
logger . error (
" Parameters are invalid. "
" Ensure architecture and size strings conform to DescribeInstanceTypes API reference. "
)
2022-10-11 15:26:03 -07:00
raise
else :
return inst_types
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.DescribeInstanceTypes]
2023-10-18 10:35:05 -07:00
2022-10-11 15:26:03 -07:00
# snippet-end:[python.example_code.ec2.InstanceWrapper.class]
