AWS Doc SDK Examples
翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
SDK for Python (Boto3) を使用する Amazon EC2の例
次のコード例は、Amazon AWS SDK for Python (Boto3) で を使用してアクションを実行し、一般的なシナリオを実装する方法を示していますEC2。
「基本」は、重要なオペレーションをサービス内で実行する方法を示すコード例です。
アクションはより大きなプログラムからのコードの抜粋であり、コンテキスト内で実行する必要があります。アクションは個々のサービス機能を呼び出す方法を示していますが、コンテキスト内のアクションは、関連するシナリオで確認できます。
「シナリオ」は、1 つのサービス内から、または他の AWS のサービスと組み合わせて複数の関数を呼び出し、特定のタスクを実行する方法を示すコード例です。
各例には、完全なソースコードへのリンクが含まれています。ここでは、コンテキストでコードを設定および実行する方法の手順を確認できます。
開始方法
次のコード例は、Amazon の使用を開始する方法を示していますEC2。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 def hello_ec2(ec2_client): """ Use the AWS SDK for Python (Boto3) to list the security groups in your account. This example uses the default settings specified in your shared credentials and config files. :param ec2_client: A Boto3 EC2 client. This client provides low-level access to AWS EC2 services. """ print("Hello, Amazon EC2! Let's list up to 10 of your security groups:") try: paginator = ec2_client.get_paginator("describe_security_groups") response_iterator = paginator.paginate(MaxResults=10) for page in response_iterator: for sg in page["SecurityGroups"]: logger.info(f"\t{sg['GroupId']}: {sg['GroupName']}") except ClientError as err: logger.error("Failed to list security groups.") if err.response["Error"]["Code"] == "AccessDeniedException": logger.error("You do not have permission to list security groups.") raise if __name__ == "__main__": hello_ec2(boto3.client("ec2"))
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeSecurityGroups「」の「」を参照してください。
-
基本
次のコードサンプルは、以下の操作方法を示しています。
キーペアとセキュリティグループを作成します。
Amazon マシンイメージ (AMI) と互換性のあるインスタンスタイプを選択し、インスタンスを作成します。
インスタンスを停止し、再起動します。
Elastic IP アドレスをインスタンスに関連付ける。
を使用してインスタンスに接続しSSH、リソースをクリーンアップします。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 コマンドプロンプトからインタラクティブのシナリオを実行します。
class EC2InstanceScenario: """ A scenario that demonstrates how to use Boto3 to manage Amazon EC2 resources. Covers creating a key pair, security group, launching an instance, associating an Elastic IP, and cleaning up resources. """ def __init__( self, inst_wrapper: EC2InstanceWrapper, key_wrapper: KeyPairWrapper, sg_wrapper: SecurityGroupWrapper, eip_wrapper: ElasticIpWrapper, ssm_client: boto3.client, remote_exec: bool = False, ): """ Initializes the EC2InstanceScenario with the necessary AWS service wrappers. :param inst_wrapper: Wrapper for EC2 instance operations. :param key_wrapper: Wrapper for key pair operations. :param sg_wrapper: Wrapper for security group operations. :param eip_wrapper: Wrapper for Elastic IP operations. :param ssm_client: Boto3 client for accessing SSM to retrieve AMIs. :param remote_exec: Flag to indicate if the scenario is running in a remote execution environment. Defaults to False. If True, the script won't prompt for user interaction. """ 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 self.remote_exec = remote_exec def create_and_list_key_pairs(self) -> None: """ Creates an RSA key pair for SSH access to the EC2 instance and lists available key pairs. """ console.print("**Step 1: Create a Secure Key Pair**", style="bold cyan") console.print( "Let's create a secure RSA key pair for connecting to your EC2 instance." ) key_name = f"MyUniqueKeyPair-{uuid.uuid4().hex[:8]}" console.print(f"- **Key Pair Name**: {key_name}") # Create the key pair and simulate the process with a progress bar. with alive_bar(1, title="Creating Key Pair") as bar: self.key_wrapper.create(key_name) time.sleep(0.4) # Simulate the delay in key creation bar() console.print(f"- **Private Key Saved to**: {self.key_wrapper.key_file_path}\n") # List key pairs (simulated) and show a progress bar. list_keys = True if list_keys: console.print("- Listing your key pairs...") start_time = time.time() with alive_bar(100, title="Listing Key Pairs") as bar: while time.time() - start_time < 2: time.sleep(0.2) bar(10) self.key_wrapper.list(5) if time.time() - start_time > 2: console.print( "Taking longer than expected! Please wait...", style="bold yellow", ) def create_security_group(self) -> None: """ Creates a security group that controls access to the EC2 instance and adds a rule to allow SSH access from the user's current public IP address. """ console.print("**Step 2: Create a Security Group**", style="bold cyan") console.print( "Security groups manage access to your instance. Let's create one." ) sg_name = f"MySecurityGroup-{uuid.uuid4().hex[:8]}" console.print(f"- **Security Group Name**: {sg_name}") # Create the security group and simulate the process with a progress bar. with alive_bar(1, title="Creating Security Group") as bar: self.sg_wrapper.create( sg_name, "Security group for example: get started with instances." ) time.sleep(0.5) bar() console.print(f"- **Security Group ID**: {self.sg_wrapper.security_group}\n") # Get the current public IP to set up SSH access. ip_response = urllib.request.urlopen("http://checkip.amazonaws.com") current_ip_address = ip_response.read().decode("utf-8").strip() console.print( "Let's add a rule to allow SSH only from your current IP address." ) console.print(f"- **Your Public IP Address**: {current_ip_address}") console.print("- Automatically adding SSH rule...") # Update security group rules to allow SSH and simulate with a progress bar. with alive_bar(1, title="Updating Security Group Rules") as bar: response = self.sg_wrapper.authorize_ingress(current_ip_address) time.sleep(0.4) if response and response.get("Return"): console.print("- **Security Group Rules Updated**.") else: console.print( "- **Error**: Couldn't update security group rules.", style="bold red", ) bar() self.sg_wrapper.describe(self.sg_wrapper.security_group) def create_instance(self) -> None: """ Launches an EC2 instance using an Amazon Linux 2 AMI and the created key pair and security group. Displays instance details and SSH connection information. """ # Retrieve Amazon Linux 2 AMIs from SSM. 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"]] ) console.print("\n**Step 3: Launch Your Instance**", style="bold cyan") console.print( "Let's create an instance from an Amazon Linux 2 AMI. Here are some options:" ) image_choice = 0 console.print(f"- Selected AMI: {amzn2_images[image_choice]['ImageId']}\n") # Display instance types compatible with the selected AMI inst_types = self.inst_wrapper.get_instance_types( amzn2_images[image_choice]["Architecture"] ) inst_type_choice = 0 console.print( f"- Selected instance type: {inst_types[inst_type_choice]['InstanceType']}\n" ) console.print("Creating your instance and waiting for it to start...") with alive_bar(1, title="Creating Instance") as bar: self.inst_wrapper.create( amzn2_images[image_choice]["ImageId"], inst_types[inst_type_choice]["InstanceType"], self.key_wrapper.key_pair["KeyName"], [self.sg_wrapper.security_group], ) time.sleep(21) bar() console.print(f"**Success! Your instance is ready:**\n", style="bold green") self.inst_wrapper.display() console.print( "You can use SSH to connect to your instance. " "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) -> None: """ Displays SSH connection information for the user to connect to the EC2 instance. Handles the case where the instance does or does not have an associated public IP address. """ if ( not self.eip_wrapper.elastic_ips or not self.eip_wrapper.elastic_ips[0].allocation_id ): if self.inst_wrapper.instances: instance = self.inst_wrapper.instances[0] instance_id = instance["InstanceId"] waiter = self.inst_wrapper.ec2_client.get_waiter("instance_running") console.print( "Waiting for the instance to be in a running state with a public IP...", style="bold cyan", ) with alive_bar(1, title="Waiting for Instance to Start") as bar: waiter.wait(InstanceIds=[instance_id]) time.sleep(20) bar() instance = self.inst_wrapper.ec2_client.describe_instances( InstanceIds=[instance_id] )["Reservations"][0]["Instances"][0] public_ip = instance.get("PublicIpAddress") if public_ip: console.print( "\nTo connect via SSH, open another command prompt and run the following command:", style="bold cyan", ) console.print( f"\tssh -i {self.key_wrapper.key_file_path} ec2-user@{public_ip}" ) else: console.print( "Instance does not have a public IP address assigned.", style="bold red", ) else: console.print( "No instance available to retrieve public IP address.", style="bold red", ) else: elastic_ip = self.eip_wrapper.elastic_ips[0] elastic_ip_address = elastic_ip.public_ip console.print( f"\tssh -i {self.key_wrapper.key_file_path} ec2-user@{elastic_ip_address}" ) if not self.remote_exec: console.print("\nOpen a new terminal tab to try the above SSH command.") input("Press Enter to continue...") def associate_elastic_ip(self) -> None: """ Allocates an Elastic IP address and associates it with the EC2 instance. Displays the Elastic IP address and SSH connection information. """ console.print("\n**Step 4: Allocate an Elastic IP Address**", style="bold cyan") console.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." ) with alive_bar(1, title="Allocating Elastic IP") as bar: elastic_ip = self.eip_wrapper.allocate() time.sleep(0.5) bar() console.print( f"- **Allocated Static Elastic IP Address**: {elastic_ip.public_ip}." ) with alive_bar(1, title="Associating Elastic IP") as bar: self.eip_wrapper.associate( elastic_ip.allocation_id, self.inst_wrapper.instances[0]["InstanceId"] ) time.sleep(2) bar() console.print(f"- **Associated Elastic IP with Your Instance**.") console.print( "You can now use SSH to connect to your instance by using the Elastic IP." ) self._display_ssh_info() def stop_and_start_instance(self) -> None: """ Stops and restarts the EC2 instance. Displays instance state and explains changes that occur when the instance is restarted, such as the potential change in the public IP address unless an Elastic IP is associated. """ console.print("\n**Step 5: Stop and Start Your Instance**", style="bold cyan") console.print("Let's stop and start your instance to see what changes.") console.print("- **Stopping your instance and waiting until it's stopped...**") with alive_bar(1, title="Stopping Instance") as bar: self.inst_wrapper.stop() time.sleep(360) bar() console.print("- **Your instance is stopped. Restarting...**") with alive_bar(1, title="Starting Instance") as bar: self.inst_wrapper.start() time.sleep(20) bar() console.print("**Your instance is running.**", style="bold green") self.inst_wrapper.display() elastic_ip = ( self.eip_wrapper.elastic_ips[0] if self.eip_wrapper.elastic_ips else None ) if elastic_ip is None or elastic_ip.allocation_id is None: console.print( "- **Note**: Every time your instance is restarted, its public IP address changes." ) else: console.print( f"Because you have associated an Elastic IP with your instance, you can \n" f"connect by using a consistent IP address after the instance restarts: {elastic_ip.public_ip}" ) self._display_ssh_info() def cleanup(self) -> None: """ Cleans up all the resources created during the scenario, including disassociating and releasing the Elastic IP, terminating the instance, deleting the security group, and deleting the key pair. """ console.print("\n**Step 6: Clean Up Resources**", style="bold cyan") console.print("Cleaning up resources:") for elastic_ip in self.eip_wrapper.elastic_ips: console.print(f"- **Elastic IP**: {elastic_ip.public_ip}") with alive_bar(1, title="Disassociating Elastic IP") as bar: self.eip_wrapper.disassociate(elastic_ip.allocation_id) time.sleep(2) bar() console.print("\t- **Disassociated Elastic IP from the Instance**") with alive_bar(1, title="Releasing Elastic IP") as bar: self.eip_wrapper.release(elastic_ip.allocation_id) time.sleep(1) bar() console.print("\t- **Released Elastic IP**") console.print(f"- **Instance**: {self.inst_wrapper.instances[0]['InstanceId']}") with alive_bar(1, title="Terminating Instance") as bar: self.inst_wrapper.terminate() time.sleep(380) bar() console.print("\t- **Terminated Instance**") console.print(f"- **Security Group**: {self.sg_wrapper.security_group}") with alive_bar(1, title="Deleting Security Group") as bar: self.sg_wrapper.delete(self.sg_wrapper.security_group) time.sleep(1) bar() console.print("\t- **Deleted Security Group**") console.print(f"- **Key Pair**: {self.key_wrapper.key_pair['KeyName']}") with alive_bar(1, title="Deleting Key Pair") as bar: self.key_wrapper.delete(self.key_wrapper.key_pair["KeyName"]) time.sleep(0.4) bar() console.print("\t- **Deleted Key Pair**") def run_scenario(self) -> None: """ Executes the entire EC2 instance scenario: creates key pairs, security groups, launches an instance, associates an Elastic IP, and cleans up all resources. """ logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") console.print("-" * 88) console.print( "Welcome to the Amazon Elastic Compute Cloud (Amazon EC2) get started with instances demo.", style="bold magenta", ) console.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() console.print("\nThanks for watching!", style="bold green") console.print("-" * 88) if __name__ == "__main__": try: scenario = EC2InstanceScenario( EC2InstanceWrapper.from_client(), KeyPairWrapper.from_client(), SecurityGroupWrapper.from_client(), ElasticIpWrapper.from_client(), boto3.client("ssm"), ) scenario.run_scenario() except Exception: logging.exception("Something went wrong with the demo.")
キーペアアクションをラップするクラスを定義します。
class KeyPairWrapper: """ Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions. This class provides methods to create, list, and delete EC2 key pairs. """ def __init__( self, ec2_client: boto3.client, key_file_dir: Union[tempfile.TemporaryDirectory, str], key_pair: Optional[dict] = None, ): """ Initializes the KeyPairWrapper with the specified EC2 client, key file directory, and an optional key pair. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A dictionary representing the Boto3 KeyPair object. This is a high-level object that wraps key pair actions. Optional. """ self.ec2_client = ec2_client self.key_pair = key_pair self.key_file_path: Optional[str] = None self.key_file_dir = key_file_dir @classmethod def from_client(cls) -> "KeyPairWrapper": """ Class method to create an instance of KeyPairWrapper using a new EC2 client and a temporary directory for storing key files. :return: An instance of KeyPairWrapper. """ ec2_client = boto3.client("ec2") return cls(ec2_client, tempfile.TemporaryDirectory()) def create(self, key_name: str) -> dict: """ 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 dictionary representing the Boto3 KeyPair object that represents the newly created key pair. :raises ClientError: If there is an error in creating the key pair, for example, if a key pair with the same name already exists. """ try: response = self.ec2_client.create_key_pair(KeyName=key_name) self.key_pair = response self.key_file_path = os.path.join( self.key_file_dir.name, f"{self.key_pair['KeyName']}.pem" ) with open(self.key_file_path, "w") as key_file: key_file.write(self.key_pair["KeyMaterial"]) except ClientError as err: if err.response["Error"]["Code"] == "InvalidKeyPair.Duplicate": logger.error( f"A key pair called {key_name} already exists. " "Please choose a different name for your key pair " "or delete the existing key pair before creating." ) raise else: return self.key_pair def list(self, limit: Optional[int] = None) -> None: """ Displays a list of key pairs for the current account. WARNING: Results are not paginated. :param limit: The maximum number of key pairs to list. If not specified, all key pairs will be listed. :raises ClientError: If there is an error in listing the key pairs. """ try: response = self.ec2_client.describe_key_pairs() key_pairs = response.get("KeyPairs", []) if limit: key_pairs = key_pairs[:limit] for key_pair in key_pairs: logger.info( f"Found {key_pair['KeyType']} key '{key_pair['KeyName']}' with fingerprint:" ) logger.info(f"\t{key_pair['KeyFingerprint']}") except ClientError as err: logger.error(f"Failed to list key pairs: {str(err)}") raise def delete(self, key_name: str) -> bool: """ Deletes a key pair by its name. :param key_name: The name of the key pair to delete. :return: A boolean indicating whether the deletion was successful. :raises ClientError: If there is an error in deleting the key pair, for example, if the key pair does not exist. """ try: self.ec2_client.delete_key_pair(KeyName=key_name) logger.info(f"Successfully deleted key pair: {key_name}") self.key_pair = None return True except self.ec2_client.exceptions.ClientError as err: logger.error(f"Deletion failed for key pair: {key_name}") error_code = err.response["Error"]["Code"] if error_code == "InvalidKeyPair.NotFound": logger.error( f"The key pair '{key_name}' does not exist and cannot be deleted. " "Please verify the key pair name and try again." ) raise
セキュリティグループのアクションをラップするクラスを定義します。
class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_client: boto3.client, security_group: Optional[str] = None): """ Initializes the SecurityGroupWrapper with an EC2 client and an optional security group ID. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param security_group: The ID of a security group to manage. This is a high-level identifier that represents the security group. """ self.ec2_client = ec2_client self.security_group = security_group @classmethod def from_client(cls) -> "SecurityGroupWrapper": """ Creates a SecurityGroupWrapper instance with a default EC2 client. :return: An instance of SecurityGroupWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def create(self, group_name: str, group_description: str) -> str: """ 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: The ID of the newly created security group. :raise Handles AWS SDK service-level ClientError, with special handling for ResourceAlreadyExists """ try: response = self.ec2_client.create_security_group( GroupName=group_name, Description=group_description ) self.security_group = response["GroupId"] except ClientError as err: if err.response["Error"]["Code"] == "ResourceAlreadyExists": logger.error( f"Security group '{group_name}' already exists. Please choose a different name." ) raise else: return self.security_group def authorize_ingress(self, ssh_ingress_ip: str) -> Optional[Dict[str, Any]]: """ 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, or None if no security group is set. :raise Handles AWS SDK service-level ClientError, with special handling for ResourceAlreadyExists """ if self.security_group is None: logger.info("No security group to update.") return None 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.ec2_client.authorize_security_group_ingress( GroupId=self.security_group, IpPermissions=ip_permissions ) except ClientError as err: if err.response["Error"]["Code"] == "InvalidPermission.Duplicate": logger.error( f"The SSH ingress rule for IP {ssh_ingress_ip} already exists" f"in security group '{self.security_group}'." ) raise else: return response def describe(self, security_group_id: Optional[str] = None) -> bool: """ Displays information about the specified security group or all security groups if no ID is provided. :param security_group_id: The ID of the security group to describe. If None, an open search is performed to describe all security groups. :returns: True if the description is successful. :raises ClientError: If there is an error describing the security group(s), such as an invalid security group ID. """ try: paginator = self.ec2_client.get_paginator("describe_security_groups") if security_group_id is None: # If no ID is provided, return all security groups. page_iterator = paginator.paginate() else: page_iterator = paginator.paginate(GroupIds=[security_group_id]) for page in page_iterator: for security_group in page["SecurityGroups"]: print(f"Security group: {security_group['GroupName']}") print(f"\tID: {security_group['GroupId']}") print(f"\tVPC: {security_group['VpcId']}") if security_group["IpPermissions"]: print("Inbound permissions:") pp(security_group["IpPermissions"]) return True except ClientError as err: logger.error("Failed to describe security group(s).") if err.response["Error"]["Code"] == "InvalidGroup.NotFound": logger.error( f"Security group {security_group_id} does not exist " f"because the specified security group ID was not found." ) raise def delete(self, security_group_id: str) -> bool: """ Deletes the specified security group. :param security_group_id: The ID of the security group to delete. Required. :returns: True if the deletion is successful. :raises ClientError: If the security group cannot be deleted due to an AWS service error. """ try: self.ec2_client.delete_security_group(GroupId=security_group_id) logger.info(f"Successfully deleted security group '{security_group_id}'") return True except ClientError as err: logger.error(f"Deletion failed for security group '{security_group_id}'") error_code = err.response["Error"]["Code"] if error_code == "InvalidGroup.NotFound": logger.error( f"Security group '{security_group_id}' cannot be deleted because it does not exist." ) elif error_code == "DependencyViolation": logger.error( f"Security group '{security_group_id}' cannot be deleted because it is still in use." " Verify that it is:" "\n\t- Detached from resources" "\n\t- Removed from references in other groups" "\n\t- Removed from VPC's as a default group" ) raise
インスタンスアクションをラップするクラスを定義します。
class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def create( self, image_id: str, instance_type: str, key_pair_name: str, security_group_ids: Optional[List[str]] = None, ) -> List[Dict[str, Any]]: """ Creates a new EC2 instance in the default VPC of the current account. The instance starts immediately after it is created. :param image_id: The ID of the Amazon Machine Image (AMI) to use for the instance. :param instance_type: The type of instance to create, such as 't2.micro'. :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. """ try: instance_params = { "ImageId": image_id, "InstanceType": instance_type, "KeyName": key_pair_name, } if security_group_ids is not None: instance_params["SecurityGroupIds"] = security_group_ids response = self.ec2_client.run_instances( **instance_params, MinCount=1, MaxCount=1 ) instance = response["Instances"][0] self.instances.append(instance) waiter = self.ec2_client.get_waiter("instance_running") waiter.wait(InstanceIds=[instance["InstanceId"]]) except ClientError as err: params_str = "\n\t".join( f"{key}: {value}" for key, value in instance_params.items() ) logger.error( f"Failed to complete instance creation request.\nRequest details:{params_str}" ) 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." ) ) raise return self.instances def display(self, state_filter: Optional[str] = "running") -> None: """ Displays information about instances, filtering by the specified state. :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'. """ if not self.instances: logger.info("No instances to display.") return instance_ids = [instance["InstanceId"] for instance in self.instances] paginator = self.ec2_client.get_paginator("describe_instances") page_iterator = paginator.paginate(InstanceIds=instance_ids) try: 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) except ClientError as err: logger.error( f"Failed to display instance(s). : {' '.join(map(str, instance_ids))}" ) 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 def terminate(self) -> None: """ Terminates instances and waits for them to reach the terminated state. """ if not self.instances: logger.info("No instances to terminate.") return instance_ids = [instance["InstanceId"] for instance in self.instances] try: 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") except ClientError as err: logger.error( f"Failed instance termination details:\n\t{str(self.instances)}" ) 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 def start(self) -> Optional[Dict[str, Any]]: """ Starts instances and waits for them to be in a running state. :return: The response to the start request. """ if not self.instances: logger.info("No instances to start.") return None instance_ids = [instance["InstanceId"] for instance in self.instances] try: 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 except ClientError as err: logger.error( f"Failed to start instance(s): {','.join(map(str, instance_ids))}" ) 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." ) raise def stop(self) -> Optional[Dict[str, Any]]: """ Stops instances and waits for them to be in a stopped state. :return: The response to the stop request, or None if there are no instances to stop. """ if not self.instances: logger.info("No instances to stop.") return None instance_ids = [instance["InstanceId"] for instance in self.instances] try: # 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) except ClientError as err: logger.error( f"Failed to stop instance(s): {','.join(map(str, instance_ids))}" ) 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." ) raise return stop_response def get_images(self, image_ids: List[str]) -> List[Dict[str, Any]]: """ Gets information about Amazon Machine Images (AMIs) from a list of AMI IDs. :param image_ids: The list of AMI IDs to look up. :return: A list of dictionaries representing the requested AMIs. """ try: response = self.ec2_client.describe_images(ImageIds=image_ids) images = response["Images"] except ClientError as err: 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.") raise return images def get_instance_types( self, architecture: str = "x86_64", sizes: List[str] = ["*.micro", "*.small"] ) -> List[Dict[str, Any]]: """ 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. """ try: inst_types = [] paginator = self.ec2_client.get_paginator("describe_instance_types") for page in paginator.paginate( Filters=[ { "Name": "processor-info.supported-architecture", "Values": [architecture], }, {"Name": "instance-type", "Values": sizes}, ] ): inst_types += page["InstanceTypes"] except ClientError as err: logger.error( f"Failed to get instance types: {architecture}, {','.join(map(str, sizes))}" ) 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." ) raise else: return inst_types
Elastic IP アクションをラップするクラスを定義します。
class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions using the client interface.""" class ElasticIp: """Represents an Elastic IP and its associated instance.""" def __init__( self, allocation_id: str, public_ip: str, instance_id: Optional[str] = None ) -> None: """ Initializes the ElasticIp object. :param allocation_id: The allocation ID of the Elastic IP. :param public_ip: The public IP address of the Elastic IP. :param instance_id: The ID of the associated EC2 instance, if any. """ self.allocation_id = allocation_id self.public_ip = public_ip self.instance_id = instance_id def __init__(self, ec2_client: Any) -> None: """ Initializes the ElasticIpWrapper with an EC2 client. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. """ self.ec2_client = ec2_client self.elastic_ips: List[ElasticIpWrapper.ElasticIp] = [] @classmethod def from_client(cls) -> "ElasticIpWrapper": """ Creates an ElasticIpWrapper instance with a default EC2 client. :return: An instance of ElasticIpWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def allocate(self) -> "ElasticIpWrapper.ElasticIp": """ 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 ElasticIp object for the newly created Elastic IP address. :raises ClientError: If the allocation fails, such as reaching the maximum limit of Elastic IPs. """ try: response = self.ec2_client.allocate_address(Domain="vpc") elastic_ip = self.ElasticIp( allocation_id=response["AllocationId"], public_ip=response["PublicIp"] ) self.elastic_ips.append(elastic_ip) except ClientError as err: if err.response["Error"]["Code"] == "AddressLimitExceeded": logger.error( "Max IP's reached. Release unused addresses or contact AWS Support for an increase." ) raise err return elastic_ip def associate( self, allocation_id: str, instance_id: str ) -> Union[Dict[str, Any], None]: """ 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 allocation_id: The allocation ID of the Elastic IP. :param instance_id: The ID of the Amazon EC2 instance. :return: A response that contains the ID of the association, or None if no Elastic IP is found. :raises ClientError: If the association fails, such as when the instance ID is not found. """ elastic_ip = self.get_elastic_ip_by_allocation(self.elastic_ips, allocation_id) if elastic_ip is None: logger.info(f"No Elastic IP found with allocation ID {allocation_id}.") return None try: response = self.ec2_client.associate_address( AllocationId=allocation_id, InstanceId=instance_id ) elastic_ip.instance_id = ( instance_id # Track the instance associated with this Elastic IP. ) except ClientError as err: if err.response["Error"]["Code"] == "InvalidInstanceID.NotFound": logger.error( f"Failed to associate Elastic IP {allocation_id} with {instance_id} " "because the specified instance ID does not exist or has not propagated fully. " "Verify the instance ID and try again, or wait a few moments before attempting to " "associate the Elastic IP address." ) raise return response def disassociate(self, allocation_id: str) -> None: """ 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. :param allocation_id: The allocation ID of the Elastic IP to disassociate. :raises ClientError: If the disassociation fails, such as when the association ID is not found. """ elastic_ip = self.get_elastic_ip_by_allocation(self.elastic_ips, allocation_id) if elastic_ip is None or elastic_ip.instance_id is None: logger.info( f"No association found for Elastic IP with allocation ID {allocation_id}." ) return try: # Retrieve the association ID before disassociating response = self.ec2_client.describe_addresses(AllocationIds=[allocation_id]) association_id = response["Addresses"][0].get("AssociationId") if association_id: self.ec2_client.disassociate_address(AssociationId=association_id) elastic_ip.instance_id = None # Remove the instance association else: logger.info( f"No Association ID found for Elastic IP with allocation ID {allocation_id}." ) except ClientError as err: if err.response["Error"]["Code"] == "InvalidAssociationID.NotFound": logger.error( f"Failed to disassociate Elastic IP {allocation_id} " "because the specified association ID for the Elastic IP address was not found. " "Verify the association ID and ensure the Elastic IP is currently associated with a " "resource before attempting to disassociate it." ) raise def release(self, allocation_id: str) -> None: """ Releases an Elastic IP address. After the Elastic IP address is released, it can no longer be used. :param allocation_id: The allocation ID of the Elastic IP to release. :raises ClientError: If the release fails, such as when the Elastic IP address is not found. """ elastic_ip = self.get_elastic_ip_by_allocation(self.elastic_ips, allocation_id) if elastic_ip is None: logger.info(f"No Elastic IP found with allocation ID {allocation_id}.") return try: self.ec2_client.release_address(AllocationId=allocation_id) self.elastic_ips.remove(elastic_ip) # Remove the Elastic IP from the list except ClientError as err: if err.response["Error"]["Code"] == "InvalidAddress.NotFound": logger.error( f"Failed to release Elastic IP address {allocation_id} " "because it could not be found. Verify the Elastic IP address " "and ensure it is allocated to your account in the correct region " "before attempting to release it." ) raise @staticmethod def get_elastic_ip_by_allocation( elastic_ips: List["ElasticIpWrapper.ElasticIp"], allocation_id: str ) -> Optional["ElasticIpWrapper.ElasticIp"]: """ Retrieves an Elastic IP object by its allocation ID from a given list of Elastic IPs. :param elastic_ips: A list of ElasticIp objects. :param allocation_id: The allocation ID of the Elastic IP to retrieve. :return: The ElasticIp object associated with the allocation ID, or None if not found. """ return next( (ip for ip in elastic_ips if ip.allocation_id == allocation_id), None )
-
API 詳細については、 AWS SDK for Python (Boto3) APIリファレンスの の以下のトピックを参照してください。
-
アクション
次のコード例は、AllocateAddress
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions using the client interface.""" class ElasticIp: """Represents an Elastic IP and its associated instance.""" def __init__( self, allocation_id: str, public_ip: str, instance_id: Optional[str] = None ) -> None: """ Initializes the ElasticIp object. :param allocation_id: The allocation ID of the Elastic IP. :param public_ip: The public IP address of the Elastic IP. :param instance_id: The ID of the associated EC2 instance, if any. """ self.allocation_id = allocation_id self.public_ip = public_ip self.instance_id = instance_id def __init__(self, ec2_client: Any) -> None: """ Initializes the ElasticIpWrapper with an EC2 client. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. """ self.ec2_client = ec2_client self.elastic_ips: List[ElasticIpWrapper.ElasticIp] = [] @classmethod def from_client(cls) -> "ElasticIpWrapper": """ Creates an ElasticIpWrapper instance with a default EC2 client. :return: An instance of ElasticIpWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def allocate(self) -> "ElasticIpWrapper.ElasticIp": """ 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 ElasticIp object for the newly created Elastic IP address. :raises ClientError: If the allocation fails, such as reaching the maximum limit of Elastic IPs. """ try: response = self.ec2_client.allocate_address(Domain="vpc") elastic_ip = self.ElasticIp( allocation_id=response["AllocationId"], public_ip=response["PublicIp"] ) self.elastic_ips.append(elastic_ip) except ClientError as err: if err.response["Error"]["Code"] == "AddressLimitExceeded": logger.error( "Max IP's reached. Release unused addresses or contact AWS Support for an increase." ) raise err return elastic_ip
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のAllocateAddress「」の「」を参照してください。
-
次のコード例は、AssociateAddress
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions using the client interface.""" class ElasticIp: """Represents an Elastic IP and its associated instance.""" def __init__( self, allocation_id: str, public_ip: str, instance_id: Optional[str] = None ) -> None: """ Initializes the ElasticIp object. :param allocation_id: The allocation ID of the Elastic IP. :param public_ip: The public IP address of the Elastic IP. :param instance_id: The ID of the associated EC2 instance, if any. """ self.allocation_id = allocation_id self.public_ip = public_ip self.instance_id = instance_id def __init__(self, ec2_client: Any) -> None: """ Initializes the ElasticIpWrapper with an EC2 client. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. """ self.ec2_client = ec2_client self.elastic_ips: List[ElasticIpWrapper.ElasticIp] = [] @classmethod def from_client(cls) -> "ElasticIpWrapper": """ Creates an ElasticIpWrapper instance with a default EC2 client. :return: An instance of ElasticIpWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def associate( self, allocation_id: str, instance_id: str ) -> Union[Dict[str, Any], None]: """ 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 allocation_id: The allocation ID of the Elastic IP. :param instance_id: The ID of the Amazon EC2 instance. :return: A response that contains the ID of the association, or None if no Elastic IP is found. :raises ClientError: If the association fails, such as when the instance ID is not found. """ elastic_ip = self.get_elastic_ip_by_allocation(self.elastic_ips, allocation_id) if elastic_ip is None: logger.info(f"No Elastic IP found with allocation ID {allocation_id}.") return None try: response = self.ec2_client.associate_address( AllocationId=allocation_id, InstanceId=instance_id ) elastic_ip.instance_id = ( instance_id # Track the instance associated with this Elastic IP. ) except ClientError as err: if err.response["Error"]["Code"] == "InvalidInstanceID.NotFound": logger.error( f"Failed to associate Elastic IP {allocation_id} with {instance_id} " "because the specified instance ID does not exist or has not propagated fully. " "Verify the instance ID and try again, or wait a few moments before attempting to " "associate the Elastic IP address." ) raise return response
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のAssociateAddress「」の「」を参照してください。
-
次の例は、AuthorizeSecurityGroupIngress
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_client: boto3.client, security_group: Optional[str] = None): """ Initializes the SecurityGroupWrapper with an EC2 client and an optional security group ID. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param security_group: The ID of a security group to manage. This is a high-level identifier that represents the security group. """ self.ec2_client = ec2_client self.security_group = security_group @classmethod def from_client(cls) -> "SecurityGroupWrapper": """ Creates a SecurityGroupWrapper instance with a default EC2 client. :return: An instance of SecurityGroupWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def authorize_ingress(self, ssh_ingress_ip: str) -> Optional[Dict[str, Any]]: """ 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, or None if no security group is set. :raise Handles AWS SDK service-level ClientError, with special handling for ResourceAlreadyExists """ if self.security_group is None: logger.info("No security group to update.") return None 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.ec2_client.authorize_security_group_ingress( GroupId=self.security_group, IpPermissions=ip_permissions ) except ClientError as err: if err.response["Error"]["Code"] == "InvalidPermission.Duplicate": logger.error( f"The SSH ingress rule for IP {ssh_ingress_ip} already exists" f"in security group '{self.security_group}'." ) raise else: return response
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のAuthorizeSecurityGroupIngress「」の「」を参照してください。
-
次の例は、CreateKeyPair
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class KeyPairWrapper: """ Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions. This class provides methods to create, list, and delete EC2 key pairs. """ def __init__( self, ec2_client: boto3.client, key_file_dir: Union[tempfile.TemporaryDirectory, str], key_pair: Optional[dict] = None, ): """ Initializes the KeyPairWrapper with the specified EC2 client, key file directory, and an optional key pair. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A dictionary representing the Boto3 KeyPair object. This is a high-level object that wraps key pair actions. Optional. """ self.ec2_client = ec2_client self.key_pair = key_pair self.key_file_path: Optional[str] = None self.key_file_dir = key_file_dir @classmethod def from_client(cls) -> "KeyPairWrapper": """ Class method to create an instance of KeyPairWrapper using a new EC2 client and a temporary directory for storing key files. :return: An instance of KeyPairWrapper. """ ec2_client = boto3.client("ec2") return cls(ec2_client, tempfile.TemporaryDirectory()) def create(self, key_name: str) -> dict: """ 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 dictionary representing the Boto3 KeyPair object that represents the newly created key pair. :raises ClientError: If there is an error in creating the key pair, for example, if a key pair with the same name already exists. """ try: response = self.ec2_client.create_key_pair(KeyName=key_name) self.key_pair = response self.key_file_path = os.path.join( self.key_file_dir.name, f"{self.key_pair['KeyName']}.pem" ) with open(self.key_file_path, "w") as key_file: key_file.write(self.key_pair["KeyMaterial"]) except ClientError as err: if err.response["Error"]["Code"] == "InvalidKeyPair.Duplicate": logger.error( f"A key pair called {key_name} already exists. " "Please choose a different name for your key pair " "or delete the existing key pair before creating." ) raise else: return self.key_pair
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のCreateKeyPair「」の「」を参照してください。
-
次の例は、CreateLaunchTemplate
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 この例では、インスタンスに特定のアクセス許可を付与するインスタンスプロファイルと、起動後にインスタンスで実行されるユーザーデータの Bash スクリプトを含む起動テンプレートを作成します。
class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def create_template( self, server_startup_script_file: str, instance_policy_file: str ) -> Dict[str, Any]: """ Creates an Amazon EC2 launch template to use with Amazon EC2 Auto Scaling. The launch template specifies a Bash script in its user data field that runs after the instance is started. This script installs Python packages and starts a Python web server on the instance. :param server_startup_script_file: The path to a Bash script file that is run when an instance starts. :param instance_policy_file: The path to a file that defines a permissions policy to create and attach to the instance profile. :return: Information about the newly created template. """ template = {} try: # Create key pair and instance profile self.create_key_pair(self.key_pair_name) self.create_instance_profile( instance_policy_file, self.instance_policy_name, self.instance_role_name, self.instance_profile_name, ) # Read the startup script with open(server_startup_script_file) as file: start_server_script = file.read() # Get the latest AMI ID ami_latest = self.ssm_client.get_parameter(Name=self.ami_param) ami_id = ami_latest["Parameter"]["Value"] # Create the launch template lt_response = self.ec2_client.create_launch_template( LaunchTemplateName=self.launch_template_name, LaunchTemplateData={ "InstanceType": self.inst_type, "ImageId": ami_id, "IamInstanceProfile": {"Name": self.instance_profile_name}, "UserData": base64.b64encode( start_server_script.encode(encoding="utf-8") ).decode(encoding="utf-8"), "KeyName": self.key_pair_name, }, ) template = lt_response["LaunchTemplate"] log.info( f"Created launch template {self.launch_template_name} for AMI {ami_id} on {self.inst_type}." ) except ClientError as err: log.error(f"Failed to create launch template {self.launch_template_name}.") error_code = err.response["Error"]["Code"] if error_code == "InvalidLaunchTemplateName.AlreadyExistsException": log.info( f"Launch template {self.launch_template_name} already exists, nothing to do." ) log.error(f"Full error:\n\t{err}") return template
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のCreateLaunchTemplate「」の「」を参照してください。
-
次のコード例は、CreateSecurityGroup
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_client: boto3.client, security_group: Optional[str] = None): """ Initializes the SecurityGroupWrapper with an EC2 client and an optional security group ID. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param security_group: The ID of a security group to manage. This is a high-level identifier that represents the security group. """ self.ec2_client = ec2_client self.security_group = security_group @classmethod def from_client(cls) -> "SecurityGroupWrapper": """ Creates a SecurityGroupWrapper instance with a default EC2 client. :return: An instance of SecurityGroupWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def create(self, group_name: str, group_description: str) -> str: """ 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: The ID of the newly created security group. :raise Handles AWS SDK service-level ClientError, with special handling for ResourceAlreadyExists """ try: response = self.ec2_client.create_security_group( GroupName=group_name, Description=group_description ) self.security_group = response["GroupId"] except ClientError as err: if err.response["Error"]["Code"] == "ResourceAlreadyExists": logger.error( f"Security group '{group_name}' already exists. Please choose a different name." ) raise else: return self.security_group
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のCreateSecurityGroup「」の「」を参照してください。
-
次のコード例は、DeleteKeyPair
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class KeyPairWrapper: """ Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions. This class provides methods to create, list, and delete EC2 key pairs. """ def __init__( self, ec2_client: boto3.client, key_file_dir: Union[tempfile.TemporaryDirectory, str], key_pair: Optional[dict] = None, ): """ Initializes the KeyPairWrapper with the specified EC2 client, key file directory, and an optional key pair. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A dictionary representing the Boto3 KeyPair object. This is a high-level object that wraps key pair actions. Optional. """ self.ec2_client = ec2_client self.key_pair = key_pair self.key_file_path: Optional[str] = None self.key_file_dir = key_file_dir @classmethod def from_client(cls) -> "KeyPairWrapper": """ Class method to create an instance of KeyPairWrapper using a new EC2 client and a temporary directory for storing key files. :return: An instance of KeyPairWrapper. """ ec2_client = boto3.client("ec2") return cls(ec2_client, tempfile.TemporaryDirectory()) def delete(self, key_name: str) -> bool: """ Deletes a key pair by its name. :param key_name: The name of the key pair to delete. :return: A boolean indicating whether the deletion was successful. :raises ClientError: If there is an error in deleting the key pair, for example, if the key pair does not exist. """ try: self.ec2_client.delete_key_pair(KeyName=key_name) logger.info(f"Successfully deleted key pair: {key_name}") self.key_pair = None return True except self.ec2_client.exceptions.ClientError as err: logger.error(f"Deletion failed for key pair: {key_name}") error_code = err.response["Error"]["Code"] if error_code == "InvalidKeyPair.NotFound": logger.error( f"The key pair '{key_name}' does not exist and cannot be deleted. " "Please verify the key pair name and try again." ) raise
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDeleteKeyPair「」の「」を参照してください。
-
次の例は、DeleteLaunchTemplate
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def delete_template(self): """ Deletes a launch template. """ try: self.ec2_client.delete_launch_template( LaunchTemplateName=self.launch_template_name ) self.delete_instance_profile( self.instance_profile_name, self.instance_role_name ) log.info("Launch template %s deleted.", self.launch_template_name) except ClientError as err: if ( err.response["Error"]["Code"] == "InvalidLaunchTemplateName.NotFoundException" ): log.info( "Launch template %s does not exist, nothing to do.", self.launch_template_name, ) log.error(f"Full error:\n\t{err}")
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDeleteLaunchTemplate「」の「」を参照してください。
-
次の例は、DeleteSecurityGroup
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_client: boto3.client, security_group: Optional[str] = None): """ Initializes the SecurityGroupWrapper with an EC2 client and an optional security group ID. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param security_group: The ID of a security group to manage. This is a high-level identifier that represents the security group. """ self.ec2_client = ec2_client self.security_group = security_group @classmethod def from_client(cls) -> "SecurityGroupWrapper": """ Creates a SecurityGroupWrapper instance with a default EC2 client. :return: An instance of SecurityGroupWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def delete(self, security_group_id: str) -> bool: """ Deletes the specified security group. :param security_group_id: The ID of the security group to delete. Required. :returns: True if the deletion is successful. :raises ClientError: If the security group cannot be deleted due to an AWS service error. """ try: self.ec2_client.delete_security_group(GroupId=security_group_id) logger.info(f"Successfully deleted security group '{security_group_id}'") return True except ClientError as err: logger.error(f"Deletion failed for security group '{security_group_id}'") error_code = err.response["Error"]["Code"] if error_code == "InvalidGroup.NotFound": logger.error( f"Security group '{security_group_id}' cannot be deleted because it does not exist." ) elif error_code == "DependencyViolation": logger.error( f"Security group '{security_group_id}' cannot be deleted because it is still in use." " Verify that it is:" "\n\t- Detached from resources" "\n\t- Removed from references in other groups" "\n\t- Removed from VPC's as a default group" ) raise
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDeleteSecurityGroup「」の「」を参照してください。
-
次の例は、DescribeAvailabilityZones
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def get_availability_zones(self) -> List[str]: """ Gets a list of Availability Zones in the AWS Region of the Amazon EC2 client. :return: The list of Availability Zones for the client Region. """ try: response = self.ec2_client.describe_availability_zones() zones = [zone["ZoneName"] for zone in response["AvailabilityZones"]] log.info(f"Retrieved {len(zones)} availability zones: {zones}.") except ClientError as err: log.error("Failed to retrieve availability zones.") log.error(f"Full error:\n\t{err}") else: return zones
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeAvailabilityZones「」の「」を参照してください。
-
次の例は、DescribeIamInstanceProfileAssociations
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def get_instance_profile(self, instance_id: str) -> Dict[str, Any]: """ Gets data about the profile associated with an instance. :param instance_id: The ID of the instance to look up. :return: The profile data. """ try: response = self.ec2_client.describe_iam_instance_profile_associations( Filters=[{"Name": "instance-id", "Values": [instance_id]}] ) if not response["IamInstanceProfileAssociations"]: log.info(f"No instance profile found for instance {instance_id}.") profile_data = response["IamInstanceProfileAssociations"][0] log.info(f"Retrieved instance profile for instance {instance_id}.") return profile_data except ClientError as err: log.error( f"Failed to retrieve instance profile for instance {instance_id}." ) error_code = err.response["Error"]["Code"] if error_code == "InvalidInstanceID.NotFound": log.error(f"The instance ID '{instance_id}' does not exist.") log.error(f"Full error:\n\t{err}")
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeIamInstanceProfileAssociations「」の「」を参照してください。
-
次のコード例は、DescribeImages
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def get_images(self, image_ids: List[str]) -> List[Dict[str, Any]]: """ Gets information about Amazon Machine Images (AMIs) from a list of AMI IDs. :param image_ids: The list of AMI IDs to look up. :return: A list of dictionaries representing the requested AMIs. """ try: response = self.ec2_client.describe_images(ImageIds=image_ids) images = response["Images"] except ClientError as err: 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.") raise return images
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeImages「」の「」を参照してください。
-
次の例は、DescribeInstanceTypes
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def get_instance_types( self, architecture: str = "x86_64", sizes: List[str] = ["*.micro", "*.small"] ) -> List[Dict[str, Any]]: """ 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. """ try: inst_types = [] paginator = self.ec2_client.get_paginator("describe_instance_types") for page in paginator.paginate( Filters=[ { "Name": "processor-info.supported-architecture", "Values": [architecture], }, {"Name": "instance-type", "Values": sizes}, ] ): inst_types += page["InstanceTypes"] except ClientError as err: logger.error( f"Failed to get instance types: {architecture}, {','.join(map(str, sizes))}" ) 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." ) raise else: return inst_types
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeInstanceTypes「」の「」を参照してください。
-
次のコード例は、DescribeInstances
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def display(self, state_filter: Optional[str] = "running") -> None: """ Displays information about instances, filtering by the specified state. :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'. """ if not self.instances: logger.info("No instances to display.") return instance_ids = [instance["InstanceId"] for instance in self.instances] paginator = self.ec2_client.get_paginator("describe_instances") page_iterator = paginator.paginate(InstanceIds=instance_ids) try: 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) except ClientError as err: logger.error( f"Failed to display instance(s). : {' '.join(map(str, instance_ids))}" ) 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
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeInstances「」の「」を参照してください。
-
次の例は、DescribeKeyPairs
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class KeyPairWrapper: """ Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) key pair actions. This class provides methods to create, list, and delete EC2 key pairs. """ def __init__( self, ec2_client: boto3.client, key_file_dir: Union[tempfile.TemporaryDirectory, str], key_pair: Optional[dict] = None, ): """ Initializes the KeyPairWrapper with the specified EC2 client, key file directory, and an optional key pair. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param key_file_dir: The folder where the private key information is stored. This should be a secure folder. :param key_pair: A dictionary representing the Boto3 KeyPair object. This is a high-level object that wraps key pair actions. Optional. """ self.ec2_client = ec2_client self.key_pair = key_pair self.key_file_path: Optional[str] = None self.key_file_dir = key_file_dir @classmethod def from_client(cls) -> "KeyPairWrapper": """ Class method to create an instance of KeyPairWrapper using a new EC2 client and a temporary directory for storing key files. :return: An instance of KeyPairWrapper. """ ec2_client = boto3.client("ec2") return cls(ec2_client, tempfile.TemporaryDirectory()) def list(self, limit: Optional[int] = None) -> None: """ Displays a list of key pairs for the current account. WARNING: Results are not paginated. :param limit: The maximum number of key pairs to list. If not specified, all key pairs will be listed. :raises ClientError: If there is an error in listing the key pairs. """ try: response = self.ec2_client.describe_key_pairs() key_pairs = response.get("KeyPairs", []) if limit: key_pairs = key_pairs[:limit] for key_pair in key_pairs: logger.info( f"Found {key_pair['KeyType']} key '{key_pair['KeyName']}' with fingerprint:" ) logger.info(f"\t{key_pair['KeyFingerprint']}") except ClientError as err: logger.error(f"Failed to list key pairs: {str(err)}") raise
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeKeyPairs「」の「」を参照してください。
-
次の例は、DescribeSecurityGroups
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class SecurityGroupWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions.""" def __init__(self, ec2_client: boto3.client, security_group: Optional[str] = None): """ Initializes the SecurityGroupWrapper with an EC2 client and an optional security group ID. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. :param security_group: The ID of a security group to manage. This is a high-level identifier that represents the security group. """ self.ec2_client = ec2_client self.security_group = security_group @classmethod def from_client(cls) -> "SecurityGroupWrapper": """ Creates a SecurityGroupWrapper instance with a default EC2 client. :return: An instance of SecurityGroupWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def describe(self, security_group_id: Optional[str] = None) -> bool: """ Displays information about the specified security group or all security groups if no ID is provided. :param security_group_id: The ID of the security group to describe. If None, an open search is performed to describe all security groups. :returns: True if the description is successful. :raises ClientError: If there is an error describing the security group(s), such as an invalid security group ID. """ try: paginator = self.ec2_client.get_paginator("describe_security_groups") if security_group_id is None: # If no ID is provided, return all security groups. page_iterator = paginator.paginate() else: page_iterator = paginator.paginate(GroupIds=[security_group_id]) for page in page_iterator: for security_group in page["SecurityGroups"]: print(f"Security group: {security_group['GroupName']}") print(f"\tID: {security_group['GroupId']}") print(f"\tVPC: {security_group['VpcId']}") if security_group["IpPermissions"]: print("Inbound permissions:") pp(security_group["IpPermissions"]) return True except ClientError as err: logger.error("Failed to describe security group(s).") if err.response["Error"]["Code"] == "InvalidGroup.NotFound": logger.error( f"Security group {security_group_id} does not exist " f"because the specified security group ID was not found." ) raise
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeSecurityGroups「」の「」を参照してください。
-
次のコード例は、DescribeSubnets
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def get_subnets(self, vpc_id: str, zones: List[str] = None) -> List[Dict[str, Any]]: """ Gets the default subnets in a VPC for a specified list of Availability Zones. :param vpc_id: The ID of the VPC to look up. :param zones: The list of Availability Zones to look up. :return: The list of subnets found. """ # Ensure that 'zones' is a list, even if None is passed if zones is None: zones = [] try: paginator = self.ec2_client.get_paginator("describe_subnets") page_iterator = paginator.paginate( Filters=[ {"Name": "vpc-id", "Values": [vpc_id]}, {"Name": "availability-zone", "Values": zones}, {"Name": "default-for-az", "Values": ["true"]}, ] ) subnets = [] for page in page_iterator: subnets.extend(page["Subnets"]) log.info("Found %s subnets for the specified zones.", len(subnets)) return subnets except ClientError as err: log.error( f"Failed to retrieve subnets for VPC '{vpc_id}' in zones {zones}." ) error_code = err.response["Error"]["Code"] if error_code == "InvalidVpcID.NotFound": log.error( "The specified VPC ID does not exist. " "Please check the VPC ID and try again." ) # Add more error-specific handling as needed log.error(f"Full error:\n\t{err}")
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeSubnets「」の「」を参照してください。
-
次のコード例は、DescribeVpcs
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def get_default_vpc(self) -> Dict[str, Any]: """ Gets the default VPC for the account. :return: Data about the default VPC. """ try: response = self.ec2_client.describe_vpcs( Filters=[{"Name": "is-default", "Values": ["true"]}] ) except ClientError as err: error_code = err.response["Error"]["Code"] log.error("Failed to retrieve the default VPC.") if error_code == "UnauthorizedOperation": log.error( "You do not have the necessary permissions to describe VPCs. " "Ensure that your AWS IAM user or role has the correct permissions." ) elif error_code == "InvalidParameterValue": log.error( "One or more parameters are invalid. Check the request parameters." ) log.error(f"Full error:\n\t{err}") else: if "Vpcs" in response and response["Vpcs"]: log.info(f"Retrieved default VPC: {response['Vpcs'][0]['VpcId']}") return response["Vpcs"][0] else: pass
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDescribeVpcs「」の「」を参照してください。
-
次の例は、DisassociateAddress
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions using the client interface.""" class ElasticIp: """Represents an Elastic IP and its associated instance.""" def __init__( self, allocation_id: str, public_ip: str, instance_id: Optional[str] = None ) -> None: """ Initializes the ElasticIp object. :param allocation_id: The allocation ID of the Elastic IP. :param public_ip: The public IP address of the Elastic IP. :param instance_id: The ID of the associated EC2 instance, if any. """ self.allocation_id = allocation_id self.public_ip = public_ip self.instance_id = instance_id def __init__(self, ec2_client: Any) -> None: """ Initializes the ElasticIpWrapper with an EC2 client. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. """ self.ec2_client = ec2_client self.elastic_ips: List[ElasticIpWrapper.ElasticIp] = [] @classmethod def from_client(cls) -> "ElasticIpWrapper": """ Creates an ElasticIpWrapper instance with a default EC2 client. :return: An instance of ElasticIpWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def disassociate(self, allocation_id: str) -> None: """ 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. :param allocation_id: The allocation ID of the Elastic IP to disassociate. :raises ClientError: If the disassociation fails, such as when the association ID is not found. """ elastic_ip = self.get_elastic_ip_by_allocation(self.elastic_ips, allocation_id) if elastic_ip is None or elastic_ip.instance_id is None: logger.info( f"No association found for Elastic IP with allocation ID {allocation_id}." ) return try: # Retrieve the association ID before disassociating response = self.ec2_client.describe_addresses(AllocationIds=[allocation_id]) association_id = response["Addresses"][0].get("AssociationId") if association_id: self.ec2_client.disassociate_address(AssociationId=association_id) elastic_ip.instance_id = None # Remove the instance association else: logger.info( f"No Association ID found for Elastic IP with allocation ID {allocation_id}." ) except ClientError as err: if err.response["Error"]["Code"] == "InvalidAssociationID.NotFound": logger.error( f"Failed to disassociate Elastic IP {allocation_id} " "because the specified association ID for the Elastic IP address was not found. " "Verify the association ID and ensure the Elastic IP is currently associated with a " "resource before attempting to disassociate it." ) raise
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のDisassociateAddress「」の「」を参照してください。
-
次のコード例は、RebootInstances
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def replace_instance_profile( self, instance_id: str, new_instance_profile_name: str, profile_association_id: str, ) -> None: """ Replaces the profile associated with a running instance. After the profile is replaced, the instance is rebooted to ensure that it uses the new profile. When the instance is ready, Systems Manager is used to restart the Python web server. :param instance_id: The ID of the instance to restart. :param new_instance_profile_name: The name of the new profile to associate with the specified instance. :param profile_association_id: The ID of the existing profile association for the instance. """ try: self.ec2_client.replace_iam_instance_profile_association( IamInstanceProfile={"Name": new_instance_profile_name}, AssociationId=profile_association_id, ) log.info( "Replaced instance profile for association %s with profile %s.", profile_association_id, new_instance_profile_name, ) time.sleep(5) self.ec2_client.reboot_instances(InstanceIds=[instance_id]) log.info("Rebooting instance %s.", instance_id) waiter = self.ec2_client.get_waiter("instance_running") log.info("Waiting for instance %s to be running.", instance_id) waiter.wait(InstanceIds=[instance_id]) log.info("Instance %s is now running.", instance_id) self.ssm_client.send_command( InstanceIds=[instance_id], DocumentName="AWS-RunShellScript", Parameters={"commands": ["cd / && sudo python3 server.py 80"]}, ) log.info(f"Restarted the Python web server on instance '{instance_id}'.") except ClientError as err: log.error("Failed to replace instance profile.") error_code = err.response["Error"]["Code"] if error_code == "InvalidAssociationID.NotFound": log.error( f"Association ID '{profile_association_id}' does not exist." "Please check the association ID and try again." ) if error_code == "InvalidInstanceId": log.error( f"The specified instance ID '{instance_id}' does not exist or is not available for SSM. " f"Please verify the instance ID and try again." ) log.error(f"Full error:\n\t{err}")
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のRebootInstances「」の「」を参照してください。
-
次の例は、ReleaseAddress
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class ElasticIpWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Elastic IP address actions using the client interface.""" class ElasticIp: """Represents an Elastic IP and its associated instance.""" def __init__( self, allocation_id: str, public_ip: str, instance_id: Optional[str] = None ) -> None: """ Initializes the ElasticIp object. :param allocation_id: The allocation ID of the Elastic IP. :param public_ip: The public IP address of the Elastic IP. :param instance_id: The ID of the associated EC2 instance, if any. """ self.allocation_id = allocation_id self.public_ip = public_ip self.instance_id = instance_id def __init__(self, ec2_client: Any) -> None: """ Initializes the ElasticIpWrapper with an EC2 client. :param ec2_client: A Boto3 Amazon EC2 client. This client provides low-level access to AWS EC2 services. """ self.ec2_client = ec2_client self.elastic_ips: List[ElasticIpWrapper.ElasticIp] = [] @classmethod def from_client(cls) -> "ElasticIpWrapper": """ Creates an ElasticIpWrapper instance with a default EC2 client. :return: An instance of ElasticIpWrapper initialized with the default EC2 client. """ ec2_client = boto3.client("ec2") return cls(ec2_client) def release(self, allocation_id: str) -> None: """ Releases an Elastic IP address. After the Elastic IP address is released, it can no longer be used. :param allocation_id: The allocation ID of the Elastic IP to release. :raises ClientError: If the release fails, such as when the Elastic IP address is not found. """ elastic_ip = self.get_elastic_ip_by_allocation(self.elastic_ips, allocation_id) if elastic_ip is None: logger.info(f"No Elastic IP found with allocation ID {allocation_id}.") return try: self.ec2_client.release_address(AllocationId=allocation_id) self.elastic_ips.remove(elastic_ip) # Remove the Elastic IP from the list except ClientError as err: if err.response["Error"]["Code"] == "InvalidAddress.NotFound": logger.error( f"Failed to release Elastic IP address {allocation_id} " "because it could not be found. Verify the Elastic IP address " "and ensure it is allocated to your account in the correct region " "before attempting to release it." ) raise
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のReleaseAddress「」の「」を参照してください。
-
次のコード例は、ReplaceIamInstanceProfileAssociation
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 この例では、実行中のインスタンスのインスタンスプロファイルを置き換え、インスタンスを再起動し、起動後にインスタンスにコマンドを送信します。
class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def replace_instance_profile( self, instance_id: str, new_instance_profile_name: str, profile_association_id: str, ) -> None: """ Replaces the profile associated with a running instance. After the profile is replaced, the instance is rebooted to ensure that it uses the new profile. When the instance is ready, Systems Manager is used to restart the Python web server. :param instance_id: The ID of the instance to restart. :param new_instance_profile_name: The name of the new profile to associate with the specified instance. :param profile_association_id: The ID of the existing profile association for the instance. """ try: self.ec2_client.replace_iam_instance_profile_association( IamInstanceProfile={"Name": new_instance_profile_name}, AssociationId=profile_association_id, ) log.info( "Replaced instance profile for association %s with profile %s.", profile_association_id, new_instance_profile_name, ) time.sleep(5) self.ec2_client.reboot_instances(InstanceIds=[instance_id]) log.info("Rebooting instance %s.", instance_id) waiter = self.ec2_client.get_waiter("instance_running") log.info("Waiting for instance %s to be running.", instance_id) waiter.wait(InstanceIds=[instance_id]) log.info("Instance %s is now running.", instance_id) self.ssm_client.send_command( InstanceIds=[instance_id], DocumentName="AWS-RunShellScript", Parameters={"commands": ["cd / && sudo python3 server.py 80"]}, ) log.info(f"Restarted the Python web server on instance '{instance_id}'.") except ClientError as err: log.error("Failed to replace instance profile.") error_code = err.response["Error"]["Code"] if error_code == "InvalidAssociationID.NotFound": log.error( f"Association ID '{profile_association_id}' does not exist." "Please check the association ID and try again." ) if error_code == "InvalidInstanceId": log.error( f"The specified instance ID '{instance_id}' does not exist or is not available for SSM. " f"Please verify the instance ID and try again." ) log.error(f"Full error:\n\t{err}")
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のReplaceIamInstanceProfileAssociation「」の「」を参照してください。
-
次の例は、RunInstances
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def create( self, image_id: str, instance_type: str, key_pair_name: str, security_group_ids: Optional[List[str]] = None, ) -> List[Dict[str, Any]]: """ Creates a new EC2 instance in the default VPC of the current account. The instance starts immediately after it is created. :param image_id: The ID of the Amazon Machine Image (AMI) to use for the instance. :param instance_type: The type of instance to create, such as 't2.micro'. :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. """ try: instance_params = { "ImageId": image_id, "InstanceType": instance_type, "KeyName": key_pair_name, } if security_group_ids is not None: instance_params["SecurityGroupIds"] = security_group_ids response = self.ec2_client.run_instances( **instance_params, MinCount=1, MaxCount=1 ) instance = response["Instances"][0] self.instances.append(instance) waiter = self.ec2_client.get_waiter("instance_running") waiter.wait(InstanceIds=[instance["InstanceId"]]) except ClientError as err: params_str = "\n\t".join( f"{key}: {value}" for key, value in instance_params.items() ) logger.error( f"Failed to complete instance creation request.\nRequest details:{params_str}" ) 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." ) ) raise return self.instances
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のRunInstances「」の「」を参照してください。
-
次の例は、StartInstances
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def start(self) -> Optional[Dict[str, Any]]: """ Starts instances and waits for them to be in a running state. :return: The response to the start request. """ if not self.instances: logger.info("No instances to start.") return None instance_ids = [instance["InstanceId"] for instance in self.instances] try: 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 except ClientError as err: logger.error( f"Failed to start instance(s): {','.join(map(str, instance_ids))}" ) 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." ) raise
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のStartInstances「」の「」を参照してください。
-
次の例は、StopInstances
を使用する方法を説明しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def stop(self) -> Optional[Dict[str, Any]]: """ Stops instances and waits for them to be in a stopped state. :return: The response to the stop request, or None if there are no instances to stop. """ if not self.instances: logger.info("No instances to stop.") return None instance_ids = [instance["InstanceId"] for instance in self.instances] try: # 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) except ClientError as err: logger.error( f"Failed to stop instance(s): {','.join(map(str, instance_ids))}" ) 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." ) raise return stop_response
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のStopInstances「」の「」を参照してください。
-
次のコード例は、TerminateInstances
を使用する方法を示しています。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 class EC2InstanceWrapper: """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) instance actions using the client interface.""" def __init__( self, ec2_client: Any, instances: Optional[List[Dict[str, Any]]] = None ) -> None: """ 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. """ self.ec2_client = ec2_client self.instances = instances or [] @classmethod 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) def terminate(self) -> None: """ Terminates instances and waits for them to reach the terminated state. """ if not self.instances: logger.info("No instances to terminate.") return instance_ids = [instance["InstanceId"] for instance in self.instances] try: 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") except ClientError as err: logger.error( f"Failed instance termination details:\n\t{str(self.instances)}" ) 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
-
API 詳細については、AWS SDKPython (Boto3) APIリファレンス のTerminateInstances「」の「」を参照してください。
-
シナリオ
次のコード例は、本、映画、曲のレコメンデーションを返す負荷分散型ウェブサービスの作成方法を示しています。この例は、障害に対するサービスの対応方法と、障害発生時の耐障害性を高めるためにサービスを再構築する方法を示しています。
Amazon EC2 Auto Scaling グループを使用して、起動テンプレートに基づいて Amazon Elastic Compute Cloud (Amazon EC2) インスタンスを作成し、インスタンスの数を指定された範囲内に保持します。
Elastic Load Balancing を使用してHTTPリクエストを処理し、配信します。 Elastic Load Balancing
Auto Scaling グループ内のインスタンスの状態を監視し、正常なインスタンスにのみリクエストを転送します。
各EC2インスタンスで Python ウェブサーバーを実行してHTTPリクエストを処理します。ウェブサーバーはレコメンデーションとヘルスチェックを返します。
Amazon DynamoDB テーブルを使用してレコメンデーションサービスをシミュレートできます。
AWS Systems Manager パラメータを更新して、リクエストとヘルスチェックに対するウェブサーバーのレスポンスを制御します。
- SDK Python 用 (Boto3)
-
注記
の詳細については、「」を参照してください GitHub。用例一覧を検索し、AWS コード例リポジトリ
での設定と実行の方法を確認してください。 コマンドプロンプトからインタラクティブのシナリオを実行します。
class Runner: """ Manages the deployment, demonstration, and destruction of resources for the resilient service. """ def __init__( self, resource_path: str, recommendation: RecommendationService, autoscaler: AutoScalingWrapper, loadbalancer: ElasticLoadBalancerWrapper, param_helper: ParameterHelper, ): """ Initializes the Runner class with the necessary parameters. :param resource_path: The path to resource files used by this example, such as IAM policies and instance scripts. :param recommendation: An instance of the RecommendationService class. :param autoscaler: An instance of the AutoScaler class. :param loadbalancer: An instance of the LoadBalancer class. :param param_helper: An instance of the ParameterHelper class. """ self.resource_path = resource_path self.recommendation = recommendation self.autoscaler = autoscaler self.loadbalancer = loadbalancer self.param_helper = param_helper self.protocol = "HTTP" self.port = 80 self.ssh_port = 22 prefix = "doc-example-resilience" self.target_group_name = f"{prefix}-tg" self.load_balancer_name = f"{prefix}-lb" def deploy(self) -> None: """ Deploys the resources required for the resilient service, including the DynamoDB table, EC2 instances, Auto Scaling group, and load balancer. """ recommendations_path = f"{self.resource_path}/recommendations.json" startup_script = f"{self.resource_path}/server_startup_script.sh" instance_policy = f"{self.resource_path}/instance_policy.json" logging.info("Starting deployment of resources for the resilient service.") logging.info( "Creating and populating DynamoDB table '%s'.", self.recommendation.table_name, ) self.recommendation.create() self.recommendation.populate(recommendations_path) logging.info( "Creating an EC2 launch template with the startup script '%s'.", startup_script, ) self.autoscaler.create_template(startup_script, instance_policy) logging.info( "Creating an EC2 Auto Scaling group across multiple Availability Zones." ) zones = self.autoscaler.create_autoscaling_group(3) logging.info("Creating variables that control the flow of the demo.") self.param_helper.reset() logging.info("Creating Elastic Load Balancing target group and load balancer.") vpc = self.autoscaler.get_default_vpc() subnets = self.autoscaler.get_subnets(vpc["VpcId"], zones) target_group = self.loadbalancer.create_target_group( self.target_group_name, self.protocol, self.port, vpc["VpcId"] ) self.loadbalancer.create_load_balancer( self.load_balancer_name, [subnet["SubnetId"] for subnet in subnets] ) self.loadbalancer.create_listener(self.load_balancer_name, target_group) self.autoscaler.attach_load_balancer_target_group(target_group) logging.info("Verifying access to the load balancer endpoint.") endpoint = self.loadbalancer.get_endpoint(self.load_balancer_name) lb_success = self.loadbalancer.verify_load_balancer_endpoint(endpoint) current_ip_address = requests.get("http://checkip.amazonaws.com").text.strip() if not lb_success: logging.warning( "Couldn't connect to the load balancer. Verifying that the port is open..." ) sec_group, port_is_open = self.autoscaler.verify_inbound_port( vpc, self.port, current_ip_address ) sec_group, ssh_port_is_open = self.autoscaler.verify_inbound_port( vpc, self.ssh_port, current_ip_address ) if not port_is_open: logging.warning( "The default security group for your VPC must allow access from this computer." ) if q.ask( f"Do you want to add a rule to security group {sec_group['GroupId']} to allow\n" f"inbound traffic on port {self.port} from your computer's IP address of {current_ip_address}? (y/n) ", q.is_yesno, ): self.autoscaler.open_inbound_port( sec_group["GroupId"], self.port, current_ip_address ) if not ssh_port_is_open: if q.ask( f"Do you want to add a rule to security group {sec_group['GroupId']} to allow\n" f"inbound SSH traffic on port {self.ssh_port} for debugging from your computer's IP address of {current_ip_address}? (y/n) ", q.is_yesno, ): self.autoscaler.open_inbound_port( sec_group["GroupId"], self.ssh_port, current_ip_address ) lb_success = self.loadbalancer.verify_load_balancer_endpoint(endpoint) if lb_success: logging.info( "Load balancer is ready. Access it at: http://%s", current_ip_address ) else: logging.error( "Couldn't get a successful response from the load balancer endpoint. Please verify your VPC and security group settings." ) def demo_choices(self) -> None: """ Presents choices for interacting with the deployed service, such as sending requests to the load balancer or checking the health of the targets. """ actions = [ "Send a GET request to the load balancer endpoint.", "Check the health of load balancer targets.", "Go to the next part of the demo.", ] choice = 0 while choice != 2: logging.info("Choose an action to interact with the service.") choice = q.choose("Which action would you like to take? ", actions) if choice == 0: logging.info("Sending a GET request to the load balancer endpoint.") endpoint = self.loadbalancer.get_endpoint(self.load_balancer_name) logging.info("GET http://%s", endpoint) response = requests.get(f"http://{endpoint}") logging.info("Response: %s", response.status_code) if response.headers.get("content-type") == "application/json": pp(response.json()) elif choice == 1: logging.info("Checking the health of load balancer targets.") health = self.loadbalancer.check_target_health(self.target_group_name) for target in health: state = target["TargetHealth"]["State"] logging.info( "Target %s on port %d is %s", target["Target"]["Id"], target["Target"]["Port"], state, ) if state != "healthy": logging.warning( "%s: %s", target["TargetHealth"]["Reason"], target["TargetHealth"]["Description"], ) logging.info( "Note that it can take a minute or two for the health check to update." ) elif choice == 2: logging.info("Proceeding to the next part of the demo.") def demo(self) -> None: """ Runs the demonstration, showing how the service responds to different failure scenarios and how a resilient architecture can keep the service running. """ ssm_only_policy = f"{self.resource_path}/ssm_only_policy.json" logging.info("Resetting parameters to starting values for the demo.") self.param_helper.reset() logging.info( "Starting demonstration of the service's resilience under various failure conditions." ) self.demo_choices() logging.info( "Simulating failure by changing the Systems Manager parameter to a non-existent table." ) self.param_helper.put(self.param_helper.table, "this-is-not-a-table") logging.info("Sending GET requests will now return failure codes.") self.demo_choices() logging.info("Switching to static response mode to mitigate failure.") self.param_helper.put(self.param_helper.failure_response, "static") logging.info("Sending GET requests will now return static responses.") self.demo_choices() logging.info("Restoring normal operation of the recommendation service.") self.param_helper.put(self.param_helper.table, self.recommendation.table_name) logging.info( "Introducing a failure by assigning bad credentials to one of the instances." ) self.autoscaler.create_instance_profile( ssm_only_policy, self.autoscaler.bad_creds_policy_name, self.autoscaler.bad_creds_role_name, self.autoscaler.bad_creds_profile_name, ["AmazonSSMManagedInstanceCore"], ) instances = self.autoscaler.get_instances() bad_instance_id = instances[0] instance_profile = self.autoscaler.get_instance_profile(bad_instance_id) logging.info( "Replacing instance profile with bad credentials for instance %s.", bad_instance_id, ) self.autoscaler.replace_instance_profile( bad_instance_id, self.autoscaler.bad_creds_profile_name, instance_profile["AssociationId"], ) logging.info( "Sending GET requests may return either a valid recommendation or a static response." ) self.demo_choices() logging.info("Implementing deep health checks to detect unhealthy instances.") self.param_helper.put(self.param_helper.health_check, "deep") logging.info("Checking the health of the load balancer targets.") self.demo_choices() logging.info( "Terminating the unhealthy instance to let the auto scaler replace it." ) self.autoscaler.terminate_instance(bad_instance_id) logging.info("The service remains resilient during instance replacement.") self.demo_choices() logging.info("Simulating a complete failure of the recommendation service.") self.param_helper.put(self.param_helper.table, "this-is-not-a-table") logging.info( "All instances will report as unhealthy, but the service will still return static responses." ) self.demo_choices() self.param_helper.reset() def destroy(self, automation=False) -> None: """ Destroys all resources created for the demo, including the load balancer, Auto Scaling group, EC2 instances, and DynamoDB table. """ logging.info( "This concludes the demo. Preparing to clean up all AWS resources created during the demo." ) if automation: cleanup = True else: cleanup = q.ask( "Do you want to clean up all demo resources? (y/n) ", q.is_yesno ) if cleanup: logging.info("Deleting load balancer and related resources.") self.loadbalancer.delete_load_balancer(self.load_balancer_name) self.loadbalancer.delete_target_group(self.target_group_name) self.autoscaler.delete_autoscaling_group(self.autoscaler.group_name) self.autoscaler.delete_key_pair() self.autoscaler.delete_template() self.autoscaler.delete_instance_profile( self.autoscaler.bad_creds_profile_name, self.autoscaler.bad_creds_role_name, ) logging.info("Deleting DynamoDB table and other resources.") self.recommendation.destroy() else: logging.warning( "Resources have not been deleted. Ensure you clean them up manually to avoid unexpected charges." ) def main() -> None: """ Main function to parse arguments and run the appropriate actions for the demo. """ parser = argparse.ArgumentParser() parser.add_argument( "--action", required=True, choices=["all", "deploy", "demo", "destroy"], help="The action to take for the demo. When 'all' is specified, resources are\n" "deployed, the demo is run, and resources are destroyed.", ) parser.add_argument( "--resource_path", default="../../../workflows/resilient_service/resources", help="The path to resource files used by this example, such as IAM policies and\n" "instance scripts.", ) args = parser.parse_args() logging.info("Starting the Resilient Service demo.") prefix = "doc-example-resilience" # Service Clients ddb_client = boto3.client("dynamodb") elb_client = boto3.client("elbv2") autoscaling_client = boto3.client("autoscaling") ec2_client = boto3.client("ec2") ssm_client = boto3.client("ssm") iam_client = boto3.client("iam") # Wrapper instantiations recommendation = RecommendationService( "doc-example-recommendation-service", ddb_client ) autoscaling_wrapper = AutoScalingWrapper( prefix, "t3.micro", "/aws/service/ami-amazon-linux-latest/amzn2-ami-hvm-x86_64-gp2", autoscaling_client, ec2_client, ssm_client, iam_client, ) elb_wrapper = ElasticLoadBalancerWrapper(elb_client) param_helper = ParameterHelper(recommendation.table_name, ssm_client) # Demo invocation runner = Runner( args.resource_path, recommendation, autoscaling_wrapper, elb_wrapper, param_helper, ) actions = [args.action] if args.action != "all" else ["deploy", "demo", "destroy"] for action in actions: if action == "deploy": runner.deploy() elif action == "demo": runner.demo() elif action == "destroy": runner.destroy() logging.info("Demo completed successfully.") if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") main()
Auto Scaling アクションと Amazon EC2アクションをラップするクラスを作成します。
class AutoScalingWrapper: """ Encapsulates Amazon EC2 Auto Scaling and EC2 management actions. """ def __init__( self, resource_prefix: str, inst_type: str, ami_param: str, autoscaling_client: boto3.client, ec2_client: boto3.client, ssm_client: boto3.client, iam_client: boto3.client, ): """ Initializes the AutoScaler class with the necessary parameters. :param resource_prefix: The prefix for naming AWS resources that are created by this class. :param inst_type: The type of EC2 instance to create, such as t3.micro. :param ami_param: The Systems Manager parameter used to look up the AMI that is created. :param autoscaling_client: A Boto3 EC2 Auto Scaling client. :param ec2_client: A Boto3 EC2 client. :param ssm_client: A Boto3 Systems Manager client. :param iam_client: A Boto3 IAM client. """ self.inst_type = inst_type self.ami_param = ami_param self.autoscaling_client = autoscaling_client self.ec2_client = ec2_client self.ssm_client = ssm_client self.iam_client = iam_client sts_client = boto3.client("sts") self.account_id = sts_client.get_caller_identity()["Account"] self.key_pair_name = f"{resource_prefix}-key-pair" self.launch_template_name = f"{resource_prefix}-template-" self.group_name = f"{resource_prefix}-group" # Happy path self.instance_policy_name = f"{resource_prefix}-pol" self.instance_role_name = f"{resource_prefix}-role" self.instance_profile_name = f"{resource_prefix}-prof" # Failure mode self.bad_creds_policy_name = f"{resource_prefix}-bc-pol" self.bad_creds_role_name = f"{resource_prefix}-bc-role" self.bad_creds_profile_name = f"{resource_prefix}-bc-prof" def create_policy(self, policy_file: str, policy_name: str) -> str: """ Creates a new IAM policy or retrieves the ARN of an existing policy. :param policy_file: The path to a JSON file that contains the policy definition. :param policy_name: The name to give the created policy. :return: The ARN of the created or existing policy. """ with open(policy_file) as file: policy_doc = file.read() try: response = self.iam_client.create_policy( PolicyName=policy_name, PolicyDocument=policy_doc ) policy_arn = response["Policy"]["Arn"] log.info(f"Policy '{policy_name}' created successfully. ARN: {policy_arn}") return policy_arn except ClientError as err: if err.response["Error"]["Code"] == "EntityAlreadyExists": # If the policy already exists, get its ARN response = self.iam_client.get_policy( PolicyArn=f"arn:aws:iam::{self.account_id}:policy/{policy_name}" ) policy_arn = response["Policy"]["Arn"] log.info(f"Policy '{policy_name}' already exists. ARN: {policy_arn}") return policy_arn log.error(f"Full error:\n\t{err}") def create_role(self, role_name: str, assume_role_doc: dict) -> str: """ Creates a new IAM role or retrieves the ARN of an existing role. :param role_name: The name to give the created role. :param assume_role_doc: The assume role policy document that specifies which entities can assume the role. :return: The ARN of the created or existing role. """ try: response = self.iam_client.create_role( RoleName=role_name, AssumeRolePolicyDocument=json.dumps(assume_role_doc) ) role_arn = response["Role"]["Arn"] log.info(f"Role '{role_name}' created successfully. ARN: {role_arn}") return role_arn except ClientError as err: if err.response["Error"]["Code"] == "EntityAlreadyExists": # If the role already exists, get its ARN response = self.iam_client.get_role(RoleName=role_name) role_arn = response["Role"]["Arn"] log.info(f"Role '{role_name}' already exists. ARN: {role_arn}") return role_arn log.error(f"Full error:\n\t{err}") def attach_policy( self, role_name: str, policy_arn: str, aws_managed_policies: Tuple[str, ...] = (), ) -> None: """ Attaches an IAM policy to a role and optionally attaches additional AWS-managed policies. :param role_name: The name of the role to attach the policy to. :param policy_arn: The ARN of the policy to attach. :param aws_managed_policies: A tuple of AWS-managed policy names to attach to the role. """ try: self.iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn) for aws_policy in aws_managed_policies: self.iam_client.attach_role_policy( RoleName=role_name, PolicyArn=f"arn:aws:iam::aws:policy/{aws_policy}", ) log.info(f"Attached policy {policy_arn} to role {role_name}.") except ClientError as err: log.error(f"Failed to attach policy {policy_arn} to role {role_name}.") log.error(f"Full error:\n\t{err}") def create_instance_profile( self, policy_file: str, policy_name: str, role_name: str, profile_name: str, aws_managed_policies: Tuple[str, ...] = (), ) -> str: """ Creates a policy, role, and profile that is associated with instances created by this class. An instance's associated profile defines a role that is assumed by the instance. The role has attached policies that specify the AWS permissions granted to clients that run on the instance. :param policy_file: The name of a JSON file that contains the policy definition to create and attach to the role. :param policy_name: The name to give the created policy. :param role_name: The name to give the created role. :param profile_name: The name to the created profile. :param aws_managed_policies: Additional AWS-managed policies that are attached to the role, such as AmazonSSMManagedInstanceCore to grant use of Systems Manager to send commands to the instance. :return: The ARN of the profile that is created. """ assume_role_doc = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "ec2.amazonaws.com"}, "Action": "sts:AssumeRole", } ], } policy_arn = self.create_policy(policy_file, policy_name) self.create_role(role_name, assume_role_doc) self.attach_policy(role_name, policy_arn, aws_managed_policies) try: profile_response = self.iam_client.create_instance_profile( InstanceProfileName=profile_name ) waiter = self.iam_client.get_waiter("instance_profile_exists") waiter.wait(InstanceProfileName=profile_name) time.sleep(10) # wait a little longer profile_arn = profile_response["InstanceProfile"]["Arn"] self.iam_client.add_role_to_instance_profile( InstanceProfileName=profile_name, RoleName=role_name ) log.info("Created profile %s and added role %s.", profile_name, role_name) except ClientError as err: if err.response["Error"]["Code"] == "EntityAlreadyExists": prof_response = self.iam_client.get_instance_profile( InstanceProfileName=profile_name ) profile_arn = prof_response["InstanceProfile"]["Arn"] log.info( "Instance profile %s already exists, nothing to do.", profile_name ) log.error(f"Full error:\n\t{err}") return profile_arn def get_instance_profile(self, instance_id: str) -> Dict[str, Any]: """ Gets data about the profile associated with an instance. :param instance_id: The ID of the instance to look up. :return: The profile data. """ try: response = self.ec2_client.describe_iam_instance_profile_associations( Filters=[{"Name": "instance-id", "Values": [instance_id]}] ) if not response["IamInstanceProfileAssociations"]: log.info(f"No instance profile found for instance {instance_id}.") profile_data = response["IamInstanceProfileAssociations"][0] log.info(f"Retrieved instance profile for instance {instance_id}.") return profile_data except ClientError as err: log.error( f"Failed to retrieve instance profile for instance {instance_id}." ) error_code = err.response["Error"]["Code"] if error_code == "InvalidInstanceID.NotFound": log.error(f"The instance ID '{instance_id}' does not exist.") log.error(f"Full error:\n\t{err}") def replace_instance_profile( self, instance_id: str, new_instance_profile_name: str, profile_association_id: str, ) -> None: """ Replaces the profile associated with a running instance. After the profile is replaced, the instance is rebooted to ensure that it uses the new profile. When the instance is ready, Systems Manager is used to restart the Python web server. :param instance_id: The ID of the instance to restart. :param new_instance_profile_name: The name of the new profile to associate with the specified instance. :param profile_association_id: The ID of the existing profile association for the instance. """ try: self.ec2_client.replace_iam_instance_profile_association( IamInstanceProfile={"Name": new_instance_profile_name}, AssociationId=profile_association_id, ) log.info( "Replaced instance profile for association %s with profile %s.", profile_association_id, new_instance_profile_name, ) time.sleep(5) self.ec2_client.reboot_instances(InstanceIds=[instance_id]) log.info("Rebooting instance %s.", instance_id) waiter = self.ec2_client.get_waiter("instance_running") log.info("Waiting for instance %s to be running.", instance_id) waiter.wait(InstanceIds=[instance_id]) log.info("Instance %s is now running.", instance_id) self.ssm_client.send_command( InstanceIds=[instance_id], DocumentName="AWS-RunShellScript", Parameters={"commands": ["cd / && sudo python3 server.py 80"]}, ) log.info(f"Restarted the Python web server on instance '{instance_id}'.") except ClientError as err: log.error("Failed to replace instance profile.") error_code = err.response["Error"]["Code"] if error_code == "InvalidAssociationID.NotFound": log.error( f"Association ID '{profile_association_id}' does not exist." "Please check the association ID and try again." ) if error_code == "InvalidInstanceId": log.error( f"The specified instance ID '{instance_id}' does not exist or is not available for SSM. " f"Please verify the instance ID and try again." ) log.error(f"Full error:\n\t{err}") def delete_instance_profile(self, profile_name: str, role_name: str) -> None: """ Detaches a role from an instance profile, detaches policies from the role, and deletes all the resources. :param profile_name: The name of the profile to delete. :param role_name: The name of the role to delete. """ try: self.iam_client.remove_role_from_instance_profile( InstanceProfileName=profile_name, RoleName=role_name ) self.iam_client.delete_instance_profile(InstanceProfileName=profile_name) log.info("Deleted instance profile %s.", profile_name) attached_policies = self.iam_client.list_attached_role_policies( RoleName=role_name ) for pol in attached_policies["AttachedPolicies"]: self.iam_client.detach_role_policy( RoleName=role_name, PolicyArn=pol["PolicyArn"] ) if not pol["PolicyArn"].startswith("arn:aws:iam::aws"): self.iam_client.delete_policy(PolicyArn=pol["PolicyArn"]) log.info("Detached and deleted policy %s.", pol["PolicyName"]) self.iam_client.delete_role(RoleName=role_name) log.info("Deleted role %s.", role_name) except ClientError as err: log.error( f"Couldn't delete instance profile {profile_name} or detach " f"policies and delete role {role_name}: {err}" ) if err.response["Error"]["Code"] == "NoSuchEntity": log.info( "Instance profile %s doesn't exist, nothing to do.", profile_name ) def create_key_pair(self, key_pair_name: str) -> None: """ Creates a new key pair. :param key_pair_name: The name of the key pair to create. """ try: response = self.ec2_client.create_key_pair(KeyName=key_pair_name) with open(f"{key_pair_name}.pem", "w") as file: file.write(response["KeyMaterial"]) chmod(f"{key_pair_name}.pem", 0o600) log.info("Created key pair %s.", key_pair_name) except ClientError as err: error_code = err.response["Error"]["Code"] log.error(f"Failed to create key pair {key_pair_name}.") if error_code == "InvalidKeyPair.Duplicate": log.error(f"A key pair with the name '{key_pair_name}' already exists.") log.error(f"Full error:\n\t{err}") def delete_key_pair(self) -> None: """ Deletes a key pair. """ try: self.ec2_client.delete_key_pair(KeyName=self.key_pair_name) remove(f"{self.key_pair_name}.pem") log.info("Deleted key pair %s.", self.key_pair_name) except ClientError as err: log.error(f"Couldn't delete key pair '{self.key_pair_name}'.") log.error(f"Full error:\n\t{err}") except FileNotFoundError as err: log.info("Key pair %s doesn't exist, nothing to do.", self.key_pair_name) log.error(f"Full error:\n\t{err}") def create_template( self, server_startup_script_file: str, instance_policy_file: str ) -> Dict[str, Any]: """ Creates an Amazon EC2 launch template to use with Amazon EC2 Auto Scaling. The launch template specifies a Bash script in its user data field that runs after the instance is started. This script installs Python packages and starts a Python web server on the instance. :param server_startup_script_file: The path to a Bash script file that is run when an instance starts. :param instance_policy_file: The path to a file that defines a permissions policy to create and attach to the instance profile. :return: Information about the newly created template. """ template = {} try: # Create key pair and instance profile self.create_key_pair(self.key_pair_name) self.create_instance_profile( instance_policy_file, self.instance_policy_name, self.instance_role_name, self.instance_profile_name, ) # Read the startup script with open(server_startup_script_file) as file: start_server_script = file.read() # Get the latest AMI ID ami_latest = self.ssm_client.get_parameter(Name=self.ami_param) ami_id = ami_latest["Parameter"]["Value"] # Create the launch template lt_response = self.ec2_client.create_launch_template( LaunchTemplateName=self.launch_template_name, LaunchTemplateData={ "InstanceType": self.inst_type, "ImageId": ami_id, "IamInstanceProfile": {"Name": self.instance_profile_name}, "UserData": base64.b64encode( start_server_script.encode(encoding="utf-8") ).decode(encoding="utf-8"), "KeyName": self.key_pair_name, }, ) template = lt_response["LaunchTemplate"] log.info( f"Created launch template {self.launch_template_name} for AMI {ami_id} on {self.inst_type}." ) except ClientError as err: log.error(f"Failed to create launch template {self.launch_template_name}.") error_code = err.response["Error"]["Code"] if error_code == "InvalidLaunchTemplateName.AlreadyExistsException": log.info( f"Launch template {self.launch_template_name} already exists, nothing to do." ) log.error(f"Full error:\n\t{err}") return template def delete_template(self): """ Deletes a launch template. """ try: self.ec2_client.delete_launch_template( LaunchTemplateName=self.launch_template_name ) self.delete_instance_profile( self.instance_profile_name, self.instance_role_name ) log.info("Launch template %s deleted.", self.launch_template_name) except ClientError as err: if ( err.response["Error"]["Code"] == "InvalidLaunchTemplateName.NotFoundException" ): log.info( "Launch template %s does not exist, nothing to do.", self.launch_template_name, ) log.error(f"Full error:\n\t{err}") def get_availability_zones(self) -> List[str]: """ Gets a list of Availability Zones in the AWS Region of the Amazon EC2 client. :return: The list of Availability Zones for the client Region. """ try: response = self.ec2_client.describe_availability_zones() zones = [zone["ZoneName"] for zone in response["AvailabilityZones"]] log.info(f"Retrieved {len(zones)} availability zones: {zones}.") except ClientError as err: log.error("Failed to retrieve availability zones.") log.error(f"Full error:\n\t{err}") else: return zones def create_autoscaling_group(self, group_size: int) -> List[str]: """ Creates an EC2 Auto Scaling group with the specified size. :param group_size: The number of instances to set for the minimum and maximum in the group. :return: The list of Availability Zones specified for the group. """ try: zones = self.get_availability_zones() self.autoscaling_client.create_auto_scaling_group( AutoScalingGroupName=self.group_name, AvailabilityZones=zones, LaunchTemplate={ "LaunchTemplateName": self.launch_template_name, "Version": "$Default", }, MinSize=group_size, MaxSize=group_size, ) log.info( f"Created EC2 Auto Scaling group {self.group_name} with availability zones {zones}." ) except ClientError as err: error_code = err.response["Error"]["Code"] if error_code == "AlreadyExists": log.info( f"EC2 Auto Scaling group {self.group_name} already exists, nothing to do." ) else: log.error(f"Failed to create EC2 Auto Scaling group {self.group_name}.") log.error(f"Full error:\n\t{err}") else: return zones def get_instances(self) -> List[str]: """ Gets data about the instances in the EC2 Auto Scaling group. :return: A list of instance IDs in the Auto Scaling group. """ try: as_response = self.autoscaling_client.describe_auto_scaling_groups( AutoScalingGroupNames=[self.group_name] ) instance_ids = [ i["InstanceId"] for i in as_response["AutoScalingGroups"][0]["Instances"] ] log.info( f"Retrieved {len(instance_ids)} instances for Auto Scaling group {self.group_name}." ) except ClientError as err: error_code = err.response["Error"]["Code"] log.error( f"Failed to retrieve instances for Auto Scaling group {self.group_name}." ) if error_code == "ResourceNotFound": log.error(f"The Auto Scaling group '{self.group_name}' does not exist.") log.error(f"Full error:\n\t{err}") else: return instance_ids def terminate_instance(self, instance_id: str, decrementsetting=False) -> None: """ Terminates an instance in an EC2 Auto Scaling group. After an instance is terminated, it can no longer be accessed. :param instance_id: The ID of the instance to terminate. :param decrementsetting: If True, do not replace terminated instances. """ try: self.autoscaling_client.terminate_instance_in_auto_scaling_group( InstanceId=instance_id, ShouldDecrementDesiredCapacity=decrementsetting, ) log.info("Terminated instance %s.", instance_id) # Adding a waiter to ensure the instance is terminated waiter = self.ec2_client.get_waiter("instance_terminated") log.info("Waiting for instance %s to be terminated...", instance_id) waiter.wait(InstanceIds=[instance_id]) log.info( f"Instance '{instance_id}' has been terminated and will be replaced." ) except ClientError as err: error_code = err.response["Error"]["Code"] log.error(f"Failed to terminate instance '{instance_id}'.") if error_code == "ScalingActivityInProgressFault": log.error( "Scaling activity is currently in progress. " "Wait for the scaling activity to complete before attempting to terminate the instance again." ) elif error_code == "ResourceContentionFault": log.error( "The request failed due to a resource contention issue. " "Ensure that no conflicting operations are being performed on the resource." ) log.error(f"Full error:\n\t{err}") def attach_load_balancer_target_group( self, lb_target_group: Dict[str, Any] ) -> None: """ Attaches an Elastic Load Balancing (ELB) target group to this EC2 Auto Scaling group. The target group specifies how the load balancer forwards requests to the instances in the group. :param lb_target_group: Data about the ELB target group to attach. """ try: self.autoscaling_client.attach_load_balancer_target_groups( AutoScalingGroupName=self.group_name, TargetGroupARNs=[lb_target_group["TargetGroupArn"]], ) log.info( "Attached load balancer target group %s to auto scaling group %s.", lb_target_group["TargetGroupName"], self.group_name, ) except ClientError as err: error_code = err.response["Error"]["Code"] log.error( f"Failed to attach load balancer target group '{lb_target_group['TargetGroupName']}'." ) if error_code == "ResourceContentionFault": log.error( "The request failed due to a resource contention issue. " "Ensure that no conflicting operations are being performed on the resource." ) elif error_code == "ServiceLinkedRoleFailure": log.error( "The operation failed because the service-linked role is not ready or does not exist. " "Check that the service-linked role exists and is correctly configured." ) log.error(f"Full error:\n\t{err}") def delete_autoscaling_group(self, group_name: str) -> None: """ Terminates all instances in the group, then deletes the EC2 Auto Scaling group. :param group_name: The name of the group to delete. """ try: response = self.autoscaling_client.describe_auto_scaling_groups( AutoScalingGroupNames=[group_name] ) groups = response.get("AutoScalingGroups", []) if len(groups) > 0: self.autoscaling_client.update_auto_scaling_group( AutoScalingGroupName=group_name, MinSize=0 ) instance_ids = [inst["InstanceId"] for inst in groups[0]["Instances"]] for inst_id in instance_ids: self.terminate_instance(inst_id) # Wait for all instances to be terminated if instance_ids: waiter = self.ec2_client.get_waiter("instance_terminated") log.info("Waiting for all instances to be terminated...") waiter.wait(InstanceIds=instance_ids) log.info("All instances have been terminated.") else: log.info(f"No groups found named '{group_name}'! Nothing to do.") except ClientError as err: error_code = err.response["Error"]["Code"] log.error(f"Failed to delete Auto Scaling group '{group_name}'.") if error_code == "ScalingActivityInProgressFault": log.error( "Scaling activity is currently in progress. " "Wait for the scaling activity to complete before attempting to delete the group again." ) elif error_code == "ResourceContentionFault": log.error( "The request failed due to a resource contention issue. " "Ensure that no conflicting operations are being performed on the group." ) log.error(f"Full error:\n\t{err}") def get_default_vpc(self) -> Dict[str, Any]: """ Gets the default VPC for the account. :return: Data about the default VPC. """ try: response = self.ec2_client.describe_vpcs( Filters=[{"Name": "is-default", "Values": ["true"]}] ) except ClientError as err: error_code = err.response["Error"]["Code"] log.error("Failed to retrieve the default VPC.") if error_code == "UnauthorizedOperation": log.error( "You do not have the necessary permissions to describe VPCs. " "Ensure that your AWS IAM user or role has the correct permissions." ) elif error_code == "InvalidParameterValue": log.error( "One or more parameters are invalid. Check the request parameters." ) log.error(f"Full error:\n\t{err}") else: if "Vpcs" in response and response["Vpcs"]: log.info(f"Retrieved default VPC: {response['Vpcs'][0]['VpcId']}") return response["Vpcs"][0] else: pass def verify_inbound_port( self, vpc: Dict[str, Any], port: int, ip_address: str ) -> Tuple[Dict[str, Any], bool]: """ Verify the default security group of the specified VPC allows ingress from this computer. This can be done by allowing ingress from this computer's IP address. In some situations, such as connecting from a corporate network, you must instead specify a prefix list ID. You can also temporarily open the port to any IP address while running this example. If you do, be sure to remove public access when you're done. :param vpc: The VPC used by this example. :param port: The port to verify. :param ip_address: This computer's IP address. :return: The default security group of the specified VPC, and a value that indicates whether the specified port is open. """ try: response = self.ec2_client.describe_security_groups( Filters=[ {"Name": "group-name", "Values": ["default"]}, {"Name": "vpc-id", "Values": [vpc["VpcId"]]}, ] ) sec_group = response["SecurityGroups"][0] port_is_open = False log.info(f"Found default security group {sec_group['GroupId']}.") for ip_perm in sec_group["IpPermissions"]: if ip_perm.get("FromPort", 0) == port: log.info(f"Found inbound rule: {ip_perm}") for ip_range in ip_perm["IpRanges"]: cidr = ip_range.get("CidrIp", "") if cidr.startswith(ip_address) or cidr == "0.0.0.0/0": port_is_open = True if ip_perm["PrefixListIds"]: port_is_open = True if not port_is_open: log.info( f"The inbound rule does not appear to be open to either this computer's IP " f"address of {ip_address}, to all IP addresses (0.0.0.0/0), or to a prefix list ID." ) else: break except ClientError as err: error_code = err.response["Error"]["Code"] log.error( f"Failed to verify inbound rule for port {port} for VPC {vpc['VpcId']}." ) if error_code == "InvalidVpcID.NotFound": log.error( f"The specified VPC ID '{vpc['VpcId']}' does not exist. Please check the VPC ID." ) log.error(f"Full error:\n\t{err}") else: return sec_group, port_is_open def open_inbound_port(self, sec_group_id: str, port: int, ip_address: str) -> None: """ Add an ingress rule to the specified security group that allows access on the specified port from the specified IP address. :param sec_group_id: The ID of the security group to modify. :param port: The port to open. :param ip_address: The IP address that is granted access. """ try: self.ec2_client.authorize_security_group_ingress( GroupId=sec_group_id, CidrIp=f"{ip_address}/32", FromPort=port, ToPort=port, IpProtocol="tcp", ) log.info( "Authorized ingress to %s on port %s from %s.", sec_group_id, port, ip_address, ) except ClientError as err: error_code = err.response["Error"]["Code"] log.error( f"Failed to authorize ingress to security group '{sec_group_id}' on port {port} from {ip_address}." ) if error_code == "InvalidGroupId.Malformed": log.error( "The security group ID is malformed. " "Please verify that the security group ID is correct." ) elif error_code == "InvalidPermission.Duplicate": log.error( "The specified rule already exists in the security group. " "Check the existing rules for this security group." ) log.error(f"Full error:\n\t{err}") def get_subnets(self, vpc_id: str, zones: List[str] = None) -> List[Dict[str, Any]]: """ Gets the default subnets in a VPC for a specified list of Availability Zones. :param vpc_id: The ID of the VPC to look up. :param zones: The list of Availability Zones to look up. :return: The list of subnets found. """ # Ensure that 'zones' is a list, even if None is passed if zones is None: zones = [] try: paginator = self.ec2_client.get_paginator("describe_subnets") page_iterator = paginator.paginate( Filters=[ {"Name": "vpc-id", "Values": [vpc_id]}, {"Name": "availability-zone", "Values": zones}, {"Name": "default-for-az", "Values": ["true"]}, ] ) subnets = [] for page in page_iterator: subnets.extend(page["Subnets"]) log.info("Found %s subnets for the specified zones.", len(subnets)) return subnets except ClientError as err: log.error( f"Failed to retrieve subnets for VPC '{vpc_id}' in zones {zones}." ) error_code = err.response["Error"]["Code"] if error_code == "InvalidVpcID.NotFound": log.error( "The specified VPC ID does not exist. " "Please check the VPC ID and try again." ) # Add more error-specific handling as needed log.error(f"Full error:\n\t{err}")
Elastic Load Balancing のアクションをラップするクラスを作成します。
class ElasticLoadBalancerWrapper: """Encapsulates Elastic Load Balancing (ELB) actions.""" def __init__(self, elb_client: boto3.client): """ Initializes the LoadBalancer class with the necessary parameters. """ self.elb_client = elb_client def create_target_group( self, target_group_name: str, protocol: str, port: int, vpc_id: str ) -> Dict[str, Any]: """ Creates an Elastic Load Balancing target group. The target group specifies how the load balancer forwards requests to instances in the group and how instance health is checked. To speed up this demo, the health check is configured with shortened times and lower thresholds. In production, you might want to decrease the sensitivity of your health checks to avoid unwanted failures. :param target_group_name: The name of the target group to create. :param protocol: The protocol to use to forward requests, such as 'HTTP'. :param port: The port to use to forward requests, such as 80. :param vpc_id: The ID of the VPC in which the load balancer exists. :return: Data about the newly created target group. """ try: response = self.elb_client.create_target_group( Name=target_group_name, Protocol=protocol, Port=port, HealthCheckPath="/healthcheck", HealthCheckIntervalSeconds=10, HealthCheckTimeoutSeconds=5, HealthyThresholdCount=2, UnhealthyThresholdCount=2, VpcId=vpc_id, ) target_group = response["TargetGroups"][0] log.info(f"Created load balancing target group '{target_group_name}'.") return target_group except ClientError as err: log.error( f"Couldn't create load balancing target group '{target_group_name}'." ) error_code = err.response["Error"]["Code"] if error_code == "DuplicateTargetGroupName": log.error( f"Target group name {target_group_name} already exists. " "Check if the target group already exists." "Consider using a different name or deleting the existing target group if appropriate." ) elif error_code == "TooManyTargetGroups": log.error( "Too many target groups exist in the account. " "Consider deleting unused target groups to create space for new ones." ) log.error(f"Full error:\n\t{err}") def delete_target_group(self, target_group_name) -> None: """ Deletes the target group. """ try: # Describe the target group to get its ARN response = self.elb_client.describe_target_groups(Names=[target_group_name]) tg_arn = response["TargetGroups"][0]["TargetGroupArn"] # Delete the target group self.elb_client.delete_target_group(TargetGroupArn=tg_arn) log.info("Deleted load balancing target group %s.", target_group_name) # Use a custom waiter to wait until the target group is no longer available self.wait_for_target_group_deletion(self.elb_client, tg_arn) log.info("Target group %s successfully deleted.", target_group_name) except ClientError as err: error_code = err.response["Error"]["Code"] log.error(f"Failed to delete target group '{target_group_name}'.") if error_code == "TargetGroupNotFound": log.error( "Load balancer target group either already deleted or never existed. " "Verify the name and check that the resource exists in the AWS Console." ) elif error_code == "ResourceInUseException": log.error( "Target group still in use by another resource. " "Ensure that the target group is no longer associated with any load balancers or resources.", ) log.error(f"Full error:\n\t{err}") def wait_for_target_group_deletion( self, elb_client, target_group_arn, max_attempts=10, delay=30 ): for attempt in range(max_attempts): try: elb_client.describe_target_groups(TargetGroupArns=[target_group_arn]) print( f"Attempt {attempt + 1}: Target group {target_group_arn} still exists." ) except ClientError as e: if e.response["Error"]["Code"] == "TargetGroupNotFound": print( f"Target group {target_group_arn} has been successfully deleted." ) return else: raise time.sleep(delay) raise TimeoutError( f"Target group {target_group_arn} was not deleted after {max_attempts * delay} seconds." ) def create_load_balancer( self, load_balancer_name: str, subnet_ids: List[str], ) -> Dict[str, Any]: """ Creates an Elastic Load Balancing load balancer that uses the specified subnets and forwards requests to the specified target group. :param load_balancer_name: The name of the load balancer to create. :param subnet_ids: A list of subnets to associate with the load balancer. :return: Data about the newly created load balancer. """ try: response = self.elb_client.create_load_balancer( Name=load_balancer_name, Subnets=subnet_ids ) load_balancer = response["LoadBalancers"][0] log.info(f"Created load balancer '{load_balancer_name}'.") waiter = self.elb_client.get_waiter("load_balancer_available") log.info( f"Waiting for load balancer '{load_balancer_name}' to be available..." ) waiter.wait(Names=[load_balancer_name]) log.info(f"Load balancer '{load_balancer_name}' is now available!") except ClientError as err: error_code = err.response["Error"]["Code"] log.error( f"Failed to create load balancer '{load_balancer_name}'. Error code: {error_code}, Message: {err.response['Error']['Message']}" ) if error_code == "DuplicateLoadBalancerNameException": log.error( f"A load balancer with the name '{load_balancer_name}' already exists. " "Load balancer names must be unique within the AWS region. " "Please choose a different name and try again." ) if error_code == "TooManyLoadBalancersException": log.error( "The maximum number of load balancers has been reached in this account and region. " "You can delete unused load balancers or request an increase in the service quota from AWS Support." ) log.error(f"Full error:\n\t{err}") else: return load_balancer def create_listener( self, load_balancer_name: str, target_group: Dict[str, Any], ) -> Dict[str, Any]: """ Creates a listener for the specified load balancer that forwards requests to the specified target group. :param load_balancer_name: The name of the load balancer to create a listener for. :param target_group: An existing target group that is added as a listener to the load balancer. :return: Data about the newly created listener. """ try: # Retrieve the load balancer ARN load_balancer_response = self.elb_client.describe_load_balancers( Names=[load_balancer_name] ) load_balancer_arn = load_balancer_response["LoadBalancers"][0][ "LoadBalancerArn" ] # Create the listener response = self.elb_client.create_listener( LoadBalancerArn=load_balancer_arn, Protocol=target_group["Protocol"], Port=target_group["Port"], DefaultActions=[ { "Type": "forward", "TargetGroupArn": target_group["TargetGroupArn"], } ], ) log.info( f"Created listener to forward traffic from load balancer '{load_balancer_name}' to target group '{target_group['TargetGroupName']}'." ) return response["Listeners"][0] except ClientError as err: error_code = err.response["Error"]["Code"] log.error( f"Failed to add a listener on '{load_balancer_name}' for target group '{target_group['TargetGroupName']}'." ) if error_code == "ListenerNotFoundException": log.error( f"The listener could not be found for the load balancer '{load_balancer_name}'. " "Please check the load balancer name and target group configuration." ) if error_code == "InvalidConfigurationRequestException": log.error( f"The configuration provided for the listener on load balancer '{load_balancer_name}' is invalid. " "Please review the provided protocol, port, and target group settings." ) log.error(f"Full error:\n\t{err}") def delete_load_balancer(self, load_balancer_name) -> None: """ Deletes a load balancer. :param load_balancer_name: The name of the load balancer to delete. """ try: response = self.elb_client.describe_load_balancers( Names=[load_balancer_name] ) lb_arn = response["LoadBalancers"][0]["LoadBalancerArn"] self.elb_client.delete_load_balancer(LoadBalancerArn=lb_arn) log.info("Deleted load balancer %s.", load_balancer_name) waiter = self.elb_client.get_waiter("load_balancers_deleted") log.info("Waiting for load balancer to be deleted...") waiter.wait(Names=[load_balancer_name]) except ClientError as err: error_code = err.response["Error"]["Code"] log.error( f"Couldn't delete load balancer '{load_balancer_name}'. Error code: {error_code}, Message: {err.response['Error']['Message']}" ) if error_code == "LoadBalancerNotFoundException": log.error( f"The load balancer '{load_balancer_name}' does not exist. " "Please check the name and try again." ) log.error(f"Full error:\n\t{err}") def get_endpoint(self, load_balancer_name) -> str: """ Gets the HTTP endpoint of the load balancer. :return: The endpoint. """ try: response = self.elb_client.describe_load_balancers( Names=[load_balancer_name] ) return response["LoadBalancers"][0]["DNSName"] except ClientError as err: log.error( f"Couldn't get the endpoint for load balancer {load_balancer_name}" ) error_code = err.response["Error"]["Code"] if error_code == "LoadBalancerNotFoundException": log.error( "Verify load balancer name and ensure it exists in the AWS console." ) log.error(f"Full error:\n\t{err}") @staticmethod def verify_load_balancer_endpoint(endpoint) -> bool: """ Verify this computer can successfully send a GET request to the load balancer endpoint. :param endpoint: The endpoint to verify. :return: True if the GET request is successful, False otherwise. """ retries = 3 verified = False while not verified and retries > 0: try: lb_response = requests.get(f"http://{endpoint}") log.info( "Got response %s from load balancer endpoint.", lb_response.status_code, ) if lb_response.status_code == 200: verified = True else: retries = 0 except requests.exceptions.ConnectionError: log.info( "Got connection error from load balancer endpoint, retrying..." ) retries -= 1 time.sleep(10) return verified def check_target_health(self, target_group_name: str) -> List[Dict[str, Any]]: """ Checks the health of the instances in the target group. :return: The health status of the target group. """ try: tg_response = self.elb_client.describe_target_groups( Names=[target_group_name] ) health_response = self.elb_client.describe_target_health( TargetGroupArn=tg_response["TargetGroups"][0]["TargetGroupArn"] ) except ClientError as err: log.error(f"Couldn't check health of {target_group_name} target(s).") error_code = err.response["Error"]["Code"] if error_code == "LoadBalancerNotFoundException": log.error( "Load balancer associated with the target group was not found. " "Ensure the load balancer exists, is in the correct AWS region, and " "that you have the necessary permissions to access it.", ) elif error_code == "TargetGroupNotFoundException": log.error( "Target group was not found. " "Verify the target group name, check that it exists in the correct region, " "and ensure it has not been deleted or created in a different account.", ) log.error(f"Full error:\n\t{err}") else: return health_response["TargetHealthDescriptions"]
DynamoDB を使用してレコメンデーションサービスをシミュレートするクラスを作成します。
class RecommendationService: """ Encapsulates a DynamoDB table to use as a service that recommends books, movies, and songs. """ def __init__(self, table_name: str, dynamodb_client: boto3.client): """ Initializes the RecommendationService class with the necessary parameters. :param table_name: The name of the DynamoDB recommendations table. :param dynamodb_client: A Boto3 DynamoDB client. """ self.table_name = table_name self.dynamodb_client = dynamodb_client def create(self) -> Dict[str, Any]: """ Creates a DynamoDB table to use as a recommendation service. The table has a hash key named 'MediaType' that defines the type of media recommended, such as Book or Movie, and a range key named 'ItemId' that, combined with the MediaType, forms a unique identifier for the recommended item. :return: Data about the newly created table. :raises RecommendationServiceError: If the table creation fails. """ try: response = self.dynamodb_client.create_table( TableName=self.table_name, AttributeDefinitions=[ {"AttributeName": "MediaType", "AttributeType": "S"}, {"AttributeName": "ItemId", "AttributeType": "N"}, ], KeySchema=[ {"AttributeName": "MediaType", "KeyType": "HASH"}, {"AttributeName": "ItemId", "KeyType": "RANGE"}, ], ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5}, ) log.info("Creating table %s...", self.table_name) waiter = self.dynamodb_client.get_waiter("table_exists") waiter.wait(TableName=self.table_name) log.info("Table %s created.", self.table_name) except ClientError as err: if err.response["Error"]["Code"] == "ResourceInUseException": log.info("Table %s exists, nothing to be done.", self.table_name) else: raise RecommendationServiceError( self.table_name, f"ClientError when creating table: {err}." ) else: return response def populate(self, data_file: str) -> None: """ Populates the recommendations table from a JSON file. :param data_file: The path to the data file. :raises RecommendationServiceError: If the table population fails. """ try: with open(data_file) as data: items = json.load(data) batch = [{"PutRequest": {"Item": item}} for item in items] self.dynamodb_client.batch_write_item(RequestItems={self.table_name: batch}) log.info( "Populated table %s with items from %s.", self.table_name, data_file ) except ClientError as err: raise RecommendationServiceError( self.table_name, f"Couldn't populate table from {data_file}: {err}" ) def destroy(self) -> None: """ Deletes the recommendations table. :raises RecommendationServiceError: If the table deletion fails. """ try: self.dynamodb_client.delete_table(TableName=self.table_name) log.info("Deleting table %s...", self.table_name) waiter = self.dynamodb_client.get_waiter("table_not_exists") waiter.wait(TableName=self.table_name) log.info("Table %s deleted.", self.table_name) except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": log.info("Table %s does not exist, nothing to do.", self.table_name) else: raise RecommendationServiceError( self.table_name, f"ClientError when deleting table: {err}." )
Systems Manager のアクションをラップするクラスを作成します。
class ParameterHelper: """ Encapsulates Systems Manager parameters. This example uses these parameters to drive the demonstration of resilient architecture, such as failure of a dependency or how the service responds to a health check. """ table: str = "doc-example-resilient-architecture-table" failure_response: str = "doc-example-resilient-architecture-failure-response" health_check: str = "doc-example-resilient-architecture-health-check" def __init__(self, table_name: str, ssm_client: boto3.client): """ Initializes the ParameterHelper class with the necessary parameters. :param table_name: The name of the DynamoDB table that is used as a recommendation service. :param ssm_client: A Boto3 Systems Manager client. """ self.ssm_client = ssm_client self.table_name = table_name def reset(self) -> None: """ Resets the Systems Manager parameters to starting values for the demo. These are the name of the DynamoDB recommendation table, no response when a dependency fails, and shallow health checks. """ self.put(self.table, self.table_name) self.put(self.failure_response, "none") self.put(self.health_check, "shallow") def put(self, name: str, value: str) -> None: """ Sets the value of a named Systems Manager parameter. :param name: The name of the parameter. :param value: The new value of the parameter. :raises ParameterHelperError: If the parameter value cannot be set. """ try: self.ssm_client.put_parameter( Name=name, Value=value, Overwrite=True, Type="String" ) log.info("Setting parameter %s to '%s'.", name, value) except ClientError as err: error_code = err.response["Error"]["Code"] log.error(f"Failed to set parameter {name}.") if error_code == "ParameterLimitExceeded": log.error( "The parameter limit has been exceeded. " "Consider deleting unused parameters or request a limit increase." ) elif error_code == "ParameterAlreadyExists": log.error( "The parameter already exists and overwrite is set to False. " "Use Overwrite=True to update the parameter." ) log.error(f"Full error:\n\t{err}")
-
API 詳細については、 AWS SDK for Python (Boto3) APIリファレンス の の以下のトピックを参照してください。
-