SDK for Python (Boto3) - AWSSDK コードサンプル

AWSDocAWS SDKGitHub サンプルリポジトリには、さらに多くの SDK サンプルがあります

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

SDK for Python (Boto3)

次のコード例は、と Amazon EC2 を使用してを使用してアクションを実行する方法を示しています。AWS SDK for Python (Boto3)

アクション」は、個々のサービス関数の呼び出し方法を示すコードの抜粋です。

シナリオ」は、同じサービス内で複数の関数を呼び出して、特定のタスクを実行する方法を示すコード例です。

それぞれの例にはGitHub、へのリンクがあり、コンテキストでコードを設定および実行する方法についての説明が記載されています。

開始方法

次のコード例は、Amazon EC2 の使用を開始する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

import boto3 def hello_ec2(ec2_resource): """ Use the AWS SDK for Python (Boto3) to create an Amazon Elastic Compute Cloud (Amazon EC2) resource and list the security groups in your account. This example uses the default settings specified in your shared credentials and config files. :param ec2_resource: A Boto3 EC2 ServiceResource object. This object is a high-level resource that wraps the low-level EC2 service API. """ print("Hello, Amazon EC2! Let's list up to 10 of your security groups:") for sg in ec2_resource.security_groups.limit(10): print(f"\t{sg.id}: {sg.group_name}") if __name__ == '__main__': hello_ec2(boto3.resource('ec2'))
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeSecurityGroups

アクション

次のコード例は、Amazon EC2 に Elastic IP アドレスを割り当てる方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions.""" def __init__(self, ec2_resource, elastic_ip=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param elastic_ip: A Boto3 VpcAddress object. This is a high-level object that wraps Elastic IP actions. """ self.ec2_resource = ec2_resource self.elastic_ip = elastic_ip @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def allocate(self): """ Allocates an Elastic IP address that can be associated with an Amazon EC2 instance. By using an Elastic IP address, you can keep the public IP address constant even when you restart the associated instance. :return: The newly created Elastic IP object. By default, the address is not associated with any instance. """ try: response = self.ec2_resource.meta.client.allocate_address(Domain='vpc') self.elastic_ip = self.ec2_resource.VpcAddress(response['AllocationId']) except ClientError as err: logger.error( "Couldn't allocate Elastic IP. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.elastic_ip
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいAllocateAddress

次のコード例は、Elastic IP アドレスを Amazon EC2 インスタンスに関連付ける方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions.""" def __init__(self, ec2_resource, elastic_ip=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param elastic_ip: A Boto3 VpcAddress object. This is a high-level object that wraps Elastic IP actions. """ self.ec2_resource = ec2_resource self.elastic_ip = elastic_ip @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def associate(self, instance): """ Associates an Elastic IP address with an instance. When this association is created, the Elastic IP's public IP address is immediately used as the public IP address of the associated instance. :param instance: A Boto3 Instance object. This is a high-level object that wraps Amazon EC2 instance actions. :return: A response that contains the ID of the association. """ if self.elastic_ip is None: logger.info("No Elastic IP to associate.") return try: response = self.elastic_ip.associate(InstanceId=instance.id) except ClientError as err: logger.error( "Couldn't associate Elastic IP %s with instance %s. Here's why: %s: %s", self.elastic_ip.allocation_id, instance.id, err.response['Error']['Code'], err.response['Error']['Message']) raise return response
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいAssociateAddress

次のコード例は、Amazon EC2 セキュリティグループを作成する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_resource, security_group=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param security_group: A Boto3 SecurityGroup object. This is a high-level object that wraps security group actions. """ self.ec2_resource = ec2_resource self.security_group = security_group @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def create(self, group_name, group_description): """ Creates a security group in the default virtual private cloud (VPC) of the current account. :param group_name: The name of the security group to create. :param group_description: The description of the security group to create. :return: A Boto3 SecurityGroup object that represents the newly created security group. """ try: self.security_group = self.ec2_resource.create_security_group( GroupName=group_name, Description=group_description) except ClientError as err: logger.error( "Couldn't create security group %s. Here's why: %s: %s", group_name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.security_group
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいCreateSecurityGroup

次のコード例は、Amazon EC2 のセキュリティkey pair を作成する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class KeyPairWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions.""" def __init__(self, ec2_resource, key_file_dir, key_pair=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A Boto3 KeyPair object. This is a high-level object that wraps key pair actions. """ self.ec2_resource = ec2_resource self.key_pair = key_pair self.key_file_path = None self.key_file_dir = key_file_dir @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource, tempfile.TemporaryDirectory()) def create(self, key_name): """ Creates a key pair that can be used to securely connect to an EC2 instance. The returned key pair contains private key information that cannot be retrieved again. The private key data is stored as a .pem file. :param key_name: The name of the key pair to create. :return: A Boto3 KeyPair object that represents the newly created key pair. """ try: self.key_pair = self.ec2_resource.create_key_pair(KeyName=key_name) self.key_file_path = os.path.join(self.key_file_dir.name, f'{self.key_pair.name}.pem') with open(self.key_file_path, 'w') as key_file: key_file.write(self.key_pair.key_material) except ClientError as err: logger.error( "Couldn't create key %s. Here's why: %s: %s", key_name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.key_pair
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいCreateKeyPair

次のコード例は、Amazon EC2 インスタンスを作成および実行する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def create( self, image, instance_type, key_pair, security_groups=None): """ Creates a new EC2 instance. The instance starts immediately after it is created. The instance is created in the default VPC of the current account. :param image: A Boto3 Image object that represents an Amazon Machine Image (AMI) that defines attributes of the instance that is created. The AMI defines things like the kind of operating system and the type of storage used by the instance. :param instance_type: The type of instance to create, such as 't2.micro'. The instance type defines things like the number of CPUs and the amount of memory. :param key_pair: A Boto3 KeyPair or KeyPairInfo object that represents the key pair that is used to secure connections to the instance. :param security_groups: A list of Boto3 SecurityGroup objects that represents the security groups that are used to grant access to the instance. When no security groups are specified, the default security group of the VPC is used. :return: A Boto3 Instance object that represents the newly created instance. """ try: instance_params = { 'ImageId': image.id, 'InstanceType': instance_type, 'KeyName': key_pair.name } if security_groups is not None: instance_params['SecurityGroupIds'] = [sg.id for sg in security_groups] self.instance = self.ec2_resource.create_instances(**instance_params, MinCount=1, MaxCount=1)[0] self.instance.wait_until_running() except ClientError as err: logging.error( "Couldn't create instance with image %s, instance type %s, and key %s. " "Here's why: %s: %s", image.id, instance_type, key_pair.name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.instance
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいRunInstances

次のコード例は、Amazon EC2 セキュリティグループを削除する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_resource, security_group=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param security_group: A Boto3 SecurityGroup object. This is a high-level object that wraps security group actions. """ self.ec2_resource = ec2_resource self.security_group = security_group @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def delete(self): """ Deletes the security group. """ if self.security_group is None: logger.info("No security group to delete.") return group_id = self.security_group.id try: self.security_group.delete() except ClientError as err: logger.error( "Couldn't delete security group %s. Here's why: %s: %s", group_id, err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDeleteSecurityGroup

次のコード例は、Amazon EC2 セキュリティkey pair を削除する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class KeyPairWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions.""" def __init__(self, ec2_resource, key_file_dir, key_pair=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A Boto3 KeyPair object. This is a high-level object that wraps key pair actions. """ self.ec2_resource = ec2_resource self.key_pair = key_pair self.key_file_path = None self.key_file_dir = key_file_dir @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource, tempfile.TemporaryDirectory()) def delete(self): """ Deletes a key pair. """ if self.key_pair is None: logger.info("No key pair to delete.") return key_name = self.key_pair.name try: self.key_pair.delete() self.key_pair = None except ClientError as err: logger.error( "Couldn't delete key %s. Here's why: %s : %s", key_name, err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDeleteKeyPair

次のコード例は、Amazon EC2 インスタンスを記述する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def display(self, indent=1): """ Displays information about an instance. :param indent: The visual indent to apply to the output. """ if self.instance is None: logger.info("No instance to display.") return try: self.instance.load() ind = '\t'*indent print(f"{ind}ID: {self.instance.id}") print(f"{ind}Image ID: {self.instance.image_id}") print(f"{ind}Instance type: {self.instance.instance_type}") print(f"{ind}Key name: {self.instance.key_name}") print(f"{ind}VPC ID: {self.instance.vpc_id}") print(f"{ind}Public IP: {self.instance.public_ip_address}") print(f"{ind}State: {self.instance.state['Name']}") except ClientError as err: logger.error( "Couldn't display your instance. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeInstances

次のコード例は、Elastic IP アドレスを Amazon EC2 Elastic IP アドレスから Elastic IP アドレスの関連付けを解除する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions.""" def __init__(self, ec2_resource, elastic_ip=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param elastic_ip: A Boto3 VpcAddress object. This is a high-level object that wraps Elastic IP actions. """ self.ec2_resource = ec2_resource self.elastic_ip = elastic_ip @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def disassociate(self): """ Removes an association between an Elastic IP address and an instance. When the association is removed, the instance is assigned a new public IP address. """ if self.elastic_ip is None: logger.info("No Elastic IP to disassociate.") return try: self.elastic_ip.association.delete() except ClientError as err: logger.error( "Couldn't disassociate Elastic IP %s from its instance. Here's why: %s: %s", self.elastic_ip.allocation_id, err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDisassociateAddress

次のコード例は、Amazon マシンイメージ (AMI) に関するデータを取得する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def get_images(self, image_ids): """ Gets information about Amazon Machine Images (AMIs) from a list of AMI IDs. :param image_ids: The list of AMIs to look up. :return: A list of Boto3 Image objects that represent the requested AMIs. """ try: images = list(self.ec2_resource.images.filter(ImageIds=image_ids)) except ClientError as err: logger.error( "Couldn't get images. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise else: return images
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeImages

次のコード例は、Amazon EC2 セキュリティグループに関するデータを取得する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_resource, security_group=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param security_group: A Boto3 SecurityGroup object. This is a high-level object that wraps security group actions. """ self.ec2_resource = ec2_resource self.security_group = security_group @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def describe(self): """ Displays information about the security group. """ if self.security_group is None: logger.info("No security group to describe.") return try: print(f"Security group: {self.security_group.group_name}") print(f"\tID: {self.security_group.id}") print(f"\tVPC: {self.security_group.vpc_id}") if self.security_group.ip_permissions: print(f"Inbound permissions:") pp(self.security_group.ip_permissions) except ClientError as err: logger.error( "Couldn't get data for security group %s. Here's why: %s: %s", self.security_group.id, err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeSecurityGroups

次のコード例は、Amazon EC2 インスタンスタイプに関するデータを取得する場合のコード例です。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def get_instance_types(self, architecture): """ Gets instance types that support the specified architecture and are designated as either 'micro' or 'small'. When an instance is created, the instance type you specify must support the architecture of the AMI you use. :param architecture: The kind of architecture the instance types must support, such as 'x86_64'. :return: A list of instance types that support the specified architecture and are either 'micro' or 'small'. """ try: inst_types = [] it_paginator = self.ec2_resource.meta.client.get_paginator('describe_instance_types') for page in it_paginator.paginate( Filters=[{ 'Name': 'processor-info.supported-architecture', 'Values': [architecture]}, {'Name': 'instance-type', 'Values': ['*.micro', '*.small']}]): inst_types += page['InstanceTypes'] except ClientError as err: logger.error( "Couldn't get instance types. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise else: return inst_types
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeInstanceTypes

次のコード例は、Amazon EC2 セキュリティキーペアを一覧表示する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class KeyPairWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions.""" def __init__(self, ec2_resource, key_file_dir, key_pair=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A Boto3 KeyPair object. This is a high-level object that wraps key pair actions. """ self.ec2_resource = ec2_resource self.key_pair = key_pair self.key_file_path = None self.key_file_dir = key_file_dir @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource, tempfile.TemporaryDirectory()) def list(self, limit): """ Displays a list of key pairs for the current account. :param limit: The maximum number of key pairs to list. """ try: for kp in self.ec2_resource.key_pairs.limit(limit): print(f"Found {kp.key_type} key {kp.name} with fingerprint:") print(f"\t{kp.key_fingerprint}") except ClientError as err: logger.error( "Couldn't list key pairs. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいDescribeKeyPairs

次のコード例は、Elastic IP アドレスを解放する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions.""" def __init__(self, ec2_resource, elastic_ip=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param elastic_ip: A Boto3 VpcAddress object. This is a high-level object that wraps Elastic IP actions. """ self.ec2_resource = ec2_resource self.elastic_ip = elastic_ip @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def release(self): """ Releases an Elastic IP address. After the Elastic IP address is released, it can no longer be used. """ if self.elastic_ip is None: logger.info("No Elastic IP to release.") return try: self.elastic_ip.release() except ClientError as err: logger.error( "Couldn't release Elastic IP address %s. Here's why: %s: %s", self.elastic_ip.allocation_id, err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいReleaseAddress

次のコード例は、Amazon EC2 セキュリティグループのインバウンドルールを設定する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_resource, security_group=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param security_group: A Boto3 SecurityGroup object. This is a high-level object that wraps security group actions. """ self.ec2_resource = ec2_resource self.security_group = security_group @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def authorize_ingress(self, ssh_ingress_ip): """ Adds a rule to the security group to allow access to SSH. :param ssh_ingress_ip: The IP address that is granted inbound access to connect to port 22 over TCP, used for SSH. :return: The response to the authorization request. The 'Return' field of the response indicates whether the request succeeded or failed. """ if self.security_group is None: logger.info("No security group to update.") return try: ip_permissions = [{ # SSH ingress open to only the specified IP address. 'IpProtocol': 'tcp', 'FromPort': 22, 'ToPort': 22, 'IpRanges': [{'CidrIp': f'{ssh_ingress_ip}/32'}]}] response = self.security_group.authorize_ingress(IpPermissions=ip_permissions) except ClientError as err: logger.error( "Couldn't authorize inbound rules for %s. Here's why: %s: %s", self.security_group.id, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいAuthorizeSecurityGroupIngress

次のコード例は、Amazon EC2 インスタンスを開始する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def start(self): """ Starts an instance and waits for it to be in a running state. :return: The response to the start request. """ if self.instance is None: logger.info("No instance to start.") return try: response = self.instance.start() self.instance.wait_until_running() except ClientError as err: logger.error( "Couldn't start instance %s. Here's why: %s: %s", self.instance.id, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいStartInstances

次のコード例は、Amazon EC2 インスタンスを停止する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def stop(self): """ Stops an instance and waits for it to be in a stopped state. :return: The response to the stop request. """ if self.instance is None: logger.info("No instance to stop.") return try: response = self.instance.stop() self.instance.wait_until_stopped() except ClientError as err: logger.error( "Couldn't stop instance %s. Here's why: %s: %s", self.instance.id, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいStopInstances

次のコード例は、Amazon EC2 インスタンスを終了する方法を示しています。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def terminate(self): """ Terminates an instance and waits for it to be in a terminated state. """ if self.instance is None: logger.info("No instance to terminate.") return instance_id = self.instance.id try: self.instance.terminate() self.instance.wait_until_terminated() self.instance = None except ClientError as err: logging.error( "Couldn't terminate instance %s. Here's why: %s: %s", instance_id, err.response['Error']['Code'], err.response['Error']['Message']) raise
  • API の詳細については、「AWSSDK for Python (Boto3) API リファレンス」のを参照してくださいTerminateInstances

シナリオ

次のコード例は、以下の操作方法を示しています。

  • key pair とセキュリティグループを作成します。

  • Amazon Machine Image Image Image (AMI) と互換性のあるインスタンスタイプを選択し、インスタンスを作成します。

  • インスタンスを停止し、再起動します。

  • Elastic IP アドレスをインスタンスに関連付ける。

  • SSH でインスタンスConnect して、リソースをクリーンアップする。

SDK for Python (Boto3)
注記

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

コマンドプロンプトからインタラクティブのシナリオを実行します。

class Ec2InstanceScenario: """Runs an interactive scenario that shows how to get started using EC2 instances.""" def __init__(self, inst_wrapper, key_wrapper, sg_wrapper, eip_wrapper, ssm_client): """ :param inst_wrapper: An object that wraps instance actions. :param key_wrapper: An object that wraps key pair actions. :param sg_wrapper: An object that wraps security group actions. :param eip_wrapper: An object that wraps Elastic IP actions. :param ssm_client: A Boto3 AWS Systems Manager client. """ self.inst_wrapper = inst_wrapper self.key_wrapper = key_wrapper self.sg_wrapper = sg_wrapper self.eip_wrapper = eip_wrapper self.ssm_client = ssm_client @demo_func def create_and_list_key_pairs(self): """ 1. Creates an RSA key pair and saves its private key data as a .pem file in secure temporary storage. The private key data is deleted after the example completes. 2. Lists the first five key pairs for the current account. """ print("Let's create an RSA key pair that you can be use to securely connect to " "your EC2 instance.") key_name = q.ask("Enter a unique name for your key: ", q.non_empty) self.key_wrapper.create(key_name) print(f"Created a key pair {self.key_wrapper.key_pair.key_name} and saved the " f"private key to {self.key_wrapper.key_file_path}.\n") if q.ask("Do you want to list some of your key pairs? (y/n) ", q.is_yesno): self.key_wrapper.list(5) @demo_func def create_security_group(self): """ 1. Creates a security group for the default VPC. 2. Adds an inbound rule to allow SSH. The SSH rule allows only inbound traffic from the current computer’s public IPv4 address. 3. Displays information about the security group. This function uses 'http://checkip.amazonaws.com' to get the current public IP address of the computer that is running the example. This method works in most cases. However, depending on how your computer connects to the internet, you might have to manually add your public IP address to the security group by using the AWS Management Console. """ print("Let's create a security group to manage access to your instance.") sg_name = q.ask("Enter a unique name for your security group: ", q.non_empty) security_group = self.sg_wrapper.create( sg_name, "Security group for example: get started with instances.") print(f"Created security group {security_group.group_name} in your default " f"VPC {security_group.vpc_id}.\n") ip_response = urllib.request.urlopen('http://checkip.amazonaws.com') current_ip_address = ip_response.read().decode('utf-8').strip() print("Let's add a rule to allow SSH only from your current IP address.") print(f"Your public IP address is {current_ip_address}.") q.ask("Press Enter to add this rule to your security group.") response = self.sg_wrapper.authorize_ingress(current_ip_address) if response['Return']: print("Security group rules updated.") else: print("Couldn't update security group rules.") self.sg_wrapper.describe() @demo_func def create_instance(self): """ 1. Gets a list of Amazon Linux 2 AMIs from AWS Systems Manager. Specifying the '/aws/service/ami-amazon-linux-latest' path returns only the latest AMIs. 2. Gets and displays information about the available AMIs and lets you select one. 3. Gets a list of instance types that are compatible with the selected AMI and lets you select one. 4. Creates an instance with the previously created key pair and security group, and the selected AMI and instance type. 5. Waits for the instance to be running and then displays its information. """ ami_paginator = self.ssm_client.get_paginator('get_parameters_by_path') ami_options = [] for page in ami_paginator.paginate(Path='/aws/service/ami-amazon-linux-latest'): ami_options += page['Parameters'] amzn2_images = self.inst_wrapper.get_images( [opt['Value'] for opt in ami_options if 'amzn2' in opt['Name']]) print("Let's create an instance from an Amazon Linux 2 AMI. Here are some options:") image_choice = q.choose( "Which one do you want to use? ", [opt.description for opt in amzn2_images]) print("Great choice!\n") print(f"Here are some instance types that support the " f"{amzn2_images[image_choice].architecture} architecture of the image:") inst_types = self.inst_wrapper.get_instance_types(amzn2_images[image_choice].architecture) inst_type_choice = q.choose( "Which one do you want to use? ", [it['InstanceType'] for it in inst_types]) print("Another great choice.\n") print("Creating your instance and waiting for it to start...") self.inst_wrapper.create( amzn2_images[image_choice], inst_types[inst_type_choice]['InstanceType'], self.key_wrapper.key_pair, [self.sg_wrapper.security_group]) print(f"Your instance is ready:\n") self.inst_wrapper.display() print("You can use SSH to connect to your instance.") print("If the connection attempt times out, you might have to manually update " "the SSH ingress rule for your IP address in the AWS Management Console.") self._display_ssh_info() def _display_ssh_info(self): """ Displays an SSH connection string that can be used to connect to a running instance. """ print("To connect, open another command prompt and run the following command:") if self.eip_wrapper.elastic_ip is None: print(f"\tssh -i {self.key_wrapper.key_file_path} " f"ec2-user@{self.inst_wrapper.instance.public_ip_address}") else: print(f"\tssh -i {self.key_wrapper.key_file_path} " f"ec2-user@{self.eip_wrapper.elastic_ip.public_ip}") q.ask("Press Enter when you're ready to continue the demo.") @demo_func def associate_elastic_ip(self): """ 1. Allocates an Elastic IP address and associates it with the instance. 2. Displays an SSH connection string that uses the Elastic IP address. """ print("You can allocate an Elastic IP address and associate it with your instance\n" "to keep a consistent IP address even when your instance restarts.") elastic_ip = self.eip_wrapper.allocate() print(f"Allocated static Elastic IP address: {elastic_ip.public_ip}.") self.eip_wrapper.associate(self.inst_wrapper.instance) print(f"Associated your Elastic IP with your instance.") print("You can now use SSH to connect to your instance by using the Elastic IP.") self._display_ssh_info() @demo_func def stop_and_start_instance(self): """ 1. Stops the instance and waits for it to stop. 2. Starts the instance and waits for it to start. 3. Displays information about the instance. 4. Displays an SSH connection string. When an Elastic IP address is associated with the instance, the IP address stays consistent when the instance stops and starts. """ print("Let's stop and start your instance to see what changes.") print("Stopping your instance and waiting until it's stopped...") self.inst_wrapper.stop() print("Your instance is stopped. Restarting...") self.inst_wrapper.start() print("Your instance is running.") self.inst_wrapper.display() if self.eip_wrapper.elastic_ip is None: print("Every time your instance is restarted, its public IP address changes.") else: print("Because you have associated an Elastic IP with your instance, you can \n" "connect by using a consistent IP address after the instance restarts.") self._display_ssh_info() @demo_func def cleanup(self): """ 1. Disassociate and delete the previously created Elastic IP. 2. Terminate the previously created instance. 3. Delete the previously created security group. 4. Delete the previously created key pair. """ print("Let's clean everything up. This example created these resources:") print(f"\tElastic IP: {self.eip_wrapper.elastic_ip.allocation_id}") print(f"\tInstance: {self.inst_wrapper.instance.id}") print(f"\tSecurity group: {self.sg_wrapper.security_group.id}") print(f"\tKey pair: {self.key_wrapper.key_pair.name}") if q.ask("Ready to delete these resources? (y/n) ", q.is_yesno): self.eip_wrapper.disassociate() print("Disassociated the Elastic IP from the instance.") self.eip_wrapper.release() print("Released the Elastic IP.") print("Terminating the instance and waiting for it to terminate...") self.inst_wrapper.terminate() print("Instance terminated.") self.sg_wrapper.delete() print("Deleted security group.") self.key_wrapper.delete() print("Deleted key pair.") def run_scenario(self): logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') print('-'*88) print("Welcome to the Amazon Elastic Compute Cloud (Amazon EC2) get started with instances demo.") print('-'*88) self.create_and_list_key_pairs() self.create_security_group() self.create_instance() self.stop_and_start_instance() self.associate_elastic_ip() self.stop_and_start_instance() self.cleanup() print("\nThanks for watching!") print('-'*88) if __name__ == '__main__': try: scenario = Ec2InstanceScenario( InstanceWrapper.from_resource(), KeyPairWrapper.from_resource(), SecurityGroupWrapper.from_resource(), ElasticIpWrapper.from_resource(), boto3.client('ssm')) scenario.run_scenario() except Exception: logging.exception("Something went wrong with the demo.")

key pair アクションをラップするクラスを定義します。

class KeyPairWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions.""" def __init__(self, ec2_resource, key_file_dir, key_pair=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A Boto3 KeyPair object. This is a high-level object that wraps key pair actions. """ self.ec2_resource = ec2_resource self.key_pair = key_pair self.key_file_path = None self.key_file_dir = key_file_dir @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource, tempfile.TemporaryDirectory()) def create(self, key_name): """ Creates a key pair that can be used to securely connect to an EC2 instance. The returned key pair contains private key information that cannot be retrieved again. The private key data is stored as a .pem file. :param key_name: The name of the key pair to create. :return: A Boto3 KeyPair object that represents the newly created key pair. """ try: self.key_pair = self.ec2_resource.create_key_pair(KeyName=key_name) self.key_file_path = os.path.join(self.key_file_dir.name, f'{self.key_pair.name}.pem') with open(self.key_file_path, 'w') as key_file: key_file.write(self.key_pair.key_material) except ClientError as err: logger.error( "Couldn't create key %s. Here's why: %s: %s", key_name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.key_pair def list(self, limit): """ Displays a list of key pairs for the current account. :param limit: The maximum number of key pairs to list. """ try: for kp in self.ec2_resource.key_pairs.limit(limit): print(f"Found {kp.key_type} key {kp.name} with fingerprint:") print(f"\t{kp.key_fingerprint}") except ClientError as err: logger.error( "Couldn't list key pairs. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise def delete(self): """ Deletes a key pair. """ if self.key_pair is None: logger.info("No key pair to delete.") return key_name = self.key_pair.name try: self.key_pair.delete() self.key_pair = None except ClientError as err: logger.error( "Couldn't delete key %s. Here's why: %s : %s", key_name, err.response['Error']['Code'], err.response['Error']['Message']) raise

セキュリティグループのアクションをラップするクラスを定義します。

class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_resource, security_group=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param security_group: A Boto3 SecurityGroup object. This is a high-level object that wraps security group actions. """ self.ec2_resource = ec2_resource self.security_group = security_group @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def create(self, group_name, group_description): """ Creates a security group in the default virtual private cloud (VPC) of the current account. :param group_name: The name of the security group to create. :param group_description: The description of the security group to create. :return: A Boto3 SecurityGroup object that represents the newly created security group. """ try: self.security_group = self.ec2_resource.create_security_group( GroupName=group_name, Description=group_description) except ClientError as err: logger.error( "Couldn't create security group %s. Here's why: %s: %s", group_name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.security_group def authorize_ingress(self, ssh_ingress_ip): """ Adds a rule to the security group to allow access to SSH. :param ssh_ingress_ip: The IP address that is granted inbound access to connect to port 22 over TCP, used for SSH. :return: The response to the authorization request. The 'Return' field of the response indicates whether the request succeeded or failed. """ if self.security_group is None: logger.info("No security group to update.") return try: ip_permissions = [{ # SSH ingress open to only the specified IP address. 'IpProtocol': 'tcp', 'FromPort': 22, 'ToPort': 22, 'IpRanges': [{'CidrIp': f'{ssh_ingress_ip}/32'}]}] response = self.security_group.authorize_ingress(IpPermissions=ip_permissions) except ClientError as err: logger.error( "Couldn't authorize inbound rules for %s. Here's why: %s: %s", self.security_group.id, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response def describe(self): """ Displays information about the security group. """ if self.security_group is None: logger.info("No security group to describe.") return try: print(f"Security group: {self.security_group.group_name}") print(f"\tID: {self.security_group.id}") print(f"\tVPC: {self.security_group.vpc_id}") if self.security_group.ip_permissions: print(f"Inbound permissions:") pp(self.security_group.ip_permissions) except ClientError as err: logger.error( "Couldn't get data for security group %s. Here's why: %s: %s", self.security_group.id, err.response['Error']['Code'], err.response['Error']['Message']) raise def delete(self): """ Deletes the security group. """ if self.security_group is None: logger.info("No security group to delete.") return group_id = self.security_group.id try: self.security_group.delete() except ClientError as err: logger.error( "Couldn't delete security group %s. Here's why: %s: %s", group_id, err.response['Error']['Code'], err.response['Error']['Message']) raise

インスタンスアクションをラップするクラスを定義します。

class InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions.""" def __init__(self, ec2_resource, instance=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param instance: A Boto3 Instance object. This is a high-level object that wraps instance actions. """ self.ec2_resource = ec2_resource self.instance = instance @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def create( self, image, instance_type, key_pair, security_groups=None): """ Creates a new EC2 instance. The instance starts immediately after it is created. The instance is created in the default VPC of the current account. :param image: A Boto3 Image object that represents an Amazon Machine Image (AMI) that defines attributes of the instance that is created. The AMI defines things like the kind of operating system and the type of storage used by the instance. :param instance_type: The type of instance to create, such as 't2.micro'. The instance type defines things like the number of CPUs and the amount of memory. :param key_pair: A Boto3 KeyPair or KeyPairInfo object that represents the key pair that is used to secure connections to the instance. :param security_groups: A list of Boto3 SecurityGroup objects that represents the security groups that are used to grant access to the instance. When no security groups are specified, the default security group of the VPC is used. :return: A Boto3 Instance object that represents the newly created instance. """ try: instance_params = { 'ImageId': image.id, 'InstanceType': instance_type, 'KeyName': key_pair.name } if security_groups is not None: instance_params['SecurityGroupIds'] = [sg.id for sg in security_groups] self.instance = self.ec2_resource.create_instances(**instance_params, MinCount=1, MaxCount=1)[0] self.instance.wait_until_running() except ClientError as err: logging.error( "Couldn't create instance with image %s, instance type %s, and key %s. " "Here's why: %s: %s", image.id, instance_type, key_pair.name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.instance def display(self, indent=1): """ Displays information about an instance. :param indent: The visual indent to apply to the output. """ if self.instance is None: logger.info("No instance to display.") return try: self.instance.load() ind = '\t'*indent print(f"{ind}ID: {self.instance.id}") print(f"{ind}Image ID: {self.instance.image_id}") print(f"{ind}Instance type: {self.instance.instance_type}") print(f"{ind}Key name: {self.instance.key_name}") print(f"{ind}VPC ID: {self.instance.vpc_id}") print(f"{ind}Public IP: {self.instance.public_ip_address}") print(f"{ind}State: {self.instance.state['Name']}") except ClientError as err: logger.error( "Couldn't display your instance. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise def terminate(self): """ Terminates an instance and waits for it to be in a terminated state. """ if self.instance is None: logger.info("No instance to terminate.") return instance_id = self.instance.id try: self.instance.terminate() self.instance.wait_until_terminated() self.instance = None except ClientError as err: logging.error( "Couldn't terminate instance %s. Here's why: %s: %s", instance_id, err.response['Error']['Code'], err.response['Error']['Message']) raise def start(self): """ Starts an instance and waits for it to be in a running state. :return: The response to the start request. """ if self.instance is None: logger.info("No instance to start.") return try: response = self.instance.start() self.instance.wait_until_running() except ClientError as err: logger.error( "Couldn't start instance %s. Here's why: %s: %s", self.instance.id, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response def stop(self): """ Stops an instance and waits for it to be in a stopped state. :return: The response to the stop request. """ if self.instance is None: logger.info("No instance to stop.") return try: response = self.instance.stop() self.instance.wait_until_stopped() except ClientError as err: logger.error( "Couldn't stop instance %s. Here's why: %s: %s", self.instance.id, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response def get_images(self, image_ids): """ Gets information about Amazon Machine Images (AMIs) from a list of AMI IDs. :param image_ids: The list of AMIs to look up. :return: A list of Boto3 Image objects that represent the requested AMIs. """ try: images = list(self.ec2_resource.images.filter(ImageIds=image_ids)) except ClientError as err: logger.error( "Couldn't get images. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise else: return images def get_instance_types(self, architecture): """ Gets instance types that support the specified architecture and are designated as either 'micro' or 'small'. When an instance is created, the instance type you specify must support the architecture of the AMI you use. :param architecture: The kind of architecture the instance types must support, such as 'x86_64'. :return: A list of instance types that support the specified architecture and are either 'micro' or 'small'. """ try: inst_types = [] it_paginator = self.ec2_resource.meta.client.get_paginator('describe_instance_types') for page in it_paginator.paginate( Filters=[{ 'Name': 'processor-info.supported-architecture', 'Values': [architecture]}, {'Name': 'instance-type', 'Values': ['*.micro', '*.small']}]): inst_types += page['InstanceTypes'] except ClientError as err: logger.error( "Couldn't get instance types. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise else: return inst_types

Elastic IP アクションをラップするクラスを定義します。

class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions.""" def __init__(self, ec2_resource, elastic_ip=None): """ :param ec2_resource: A Boto3 Amazon EC2 resource. This high-level resource is used to create additional high-level objects that wrap low-level Amazon EC2 service actions. :param elastic_ip: A Boto3 VpcAddress object. This is a high-level object that wraps Elastic IP actions. """ self.ec2_resource = ec2_resource self.elastic_ip = elastic_ip @classmethod def from_resource(cls): ec2_resource = boto3.resource('ec2') return cls(ec2_resource) def allocate(self): """ Allocates an Elastic IP address that can be associated with an Amazon EC2 instance. By using an Elastic IP address, you can keep the public IP address constant even when you restart the associated instance. :return: The newly created Elastic IP object. By default, the address is not associated with any instance. """ try: response = self.ec2_resource.meta.client.allocate_address(Domain='vpc') self.elastic_ip = self.ec2_resource.VpcAddress(response['AllocationId']) except ClientError as err: logger.error( "Couldn't allocate Elastic IP. Here's why: %s: %s", err.response['Error']['Code'], err.response['Error']['Message']) raise else: return self.elastic_ip def associate(self, instance): """ Associates an Elastic IP address with an instance. When this association is created, the Elastic IP's public IP address is immediately used as the public IP address of the associated instance. :param instance: A Boto3 Instance object. This is a high-level object that wraps Amazon EC2 instance actions. :return: A response that contains the ID of the association. """ if self.elastic_ip is None: logger.info("No Elastic IP to associate.") return try: response = self.elastic_ip.associate(InstanceId=instance.id) except ClientError as err: logger.error( "Couldn't associate Elastic IP %s with instance %s. Here's why: %s: %s", self.elastic_ip.allocation_id, instance.id, err.response['Error']['Code'], err.response['Error']['Message']) raise return response def disassociate(self): """ Removes an association between an Elastic IP address and an instance. When the association is removed, the instance is assigned a new public IP address. """ if self.elastic_ip is None: logger.info("No Elastic IP to disassociate.") return try: self.elastic_ip.association.delete() except ClientError as err: logger.error( "Couldn't disassociate Elastic IP %s from its instance. Here's why: %s: %s", self.elastic_ip.allocation_id, err.response['Error']['Code'], err.response['Error']['Message']) raise def release(self): """ Releases an Elastic IP address. After the Elastic IP address is released, it can no longer be used. """ if self.elastic_ip is None: logger.info("No Elastic IP to release.") return try: self.elastic_ip.release() except ClientError as err: logger.error( "Couldn't release Elastic IP address %s. Here's why: %s: %s", self.elastic_ip.allocation_id, err.response['Error']['Code'], err.response['Error']['Message']) raise