Welcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.
|
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
||
|
|
# SPDX-License-Identifier: Apache-2.0
|
||
|
|
import logging
|
||
|
|
from pprint import pp
|
||
|
|
from typing import Any, Dict
|
||
|
|
|
||
|
|
import boto3
|
||
|
|
from botocore.exceptions import ClientError, ValidationError
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ec2.VpcWrapper.class]
|
||
|
|
# snippet-start:[python.example_code.ec2.VpcWrapper.decl]
|
||
|
|
class VpcWrapper:
|
||
|
|
"""Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) Amazon Virtual Private Cloud actions."""
|
||
|
|
|
||
|
|
def __init__(self, ec2_client: boto3.client):
|
||
|
|
"""
|
||
|
|
Initializes the VpcWrapper 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
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_client(cls) -> "VpcWrapper":
|
||
|
|
"""
|
||
|
|
Creates a VpcWrapper instance with a default EC2 client.
|
||
|
|
|
||
|
|
:return: An instance of VpcWrapper initialized with the default EC2 client.
|
||
|
|
"""
|
||
|
|
ec2_client = boto3.client("ec2")
|
||
|
|
return cls(ec2_client)
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ec2.VpcWrapper.decl]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ec2.CreateVpc]
|
||
|
|
def create(self, cidr_block: str) -> str:
|
||
|
|
"""
|
||
|
|
Creates a new Amazon VPC with the specified CIDR block.
|
||
|
|
|
||
|
|
:param cidr_block: The CIDR block for the new VPC, such as '10.0.0.0/16'.
|
||
|
|
:return: The ID of the new VPC.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
response = self.ec2_client.create_vpc(CidrBlock=cidr_block)
|
||
|
|
vpc_id = response["Vpc"]["VpcId"]
|
||
|
|
|
||
|
|
waiter = self.ec2_client.get_waiter("vpc_available")
|
||
|
|
waiter.wait(VpcIds=[vpc_id])
|
||
|
|
return vpc_id
|
||
|
|
except ClientError as client_error:
|
||
|
|
logging.error(
|
||
|
|
"Couldn't create the vpc. Here's why: %s",
|
||
|
|
client_error.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ec2.CreateVpc]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ec2.DescribeRouteTables]
|
||
|
|
def describe_route_tables(self, vpc_ids: list[str]) -> None:
|
||
|
|
"""
|
||
|
|
Displays information about the route tables in the specified VPC.
|
||
|
|
|
||
|
|
:param vpc_ids: A list of VPC IDs.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
response = self.ec2_client.describe_route_tables(
|
||
|
|
Filters=[{"Name": "vpc-id", "Values": vpc_ids}]
|
||
|
|
)
|
||
|
|
pp(response["RouteTables"])
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't describe route tables for VPCs %s. Here's why: %s: %s",
|
||
|
|
vpc_ids,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ec2.DescribeRouteTables]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ec2.CreateVpcEndpoint]
|
||
|
|
def create_vpc_endpoint(
|
||
|
|
self, vpc_id: str, service_name: str, route_table_ids: list[str]
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""
|
||
|
|
Creates a new VPC endpoint for the specified service and associates it with the specified route tables.
|
||
|
|
|
||
|
|
:param vpc_id: The ID of the VPC to create the endpoint in.
|
||
|
|
:param service_name: The name of the service to create the endpoint for.
|
||
|
|
:param route_table_ids: A list of IDs of the route tables to associate with the endpoint.
|
||
|
|
:return: A dictionary representing the newly created VPC endpoint.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
response = self.ec2_client.create_vpc_endpoint(
|
||
|
|
VpcId=vpc_id,
|
||
|
|
ServiceName=service_name,
|
||
|
|
RouteTableIds=route_table_ids,
|
||
|
|
)
|
||
|
|
return response["VpcEndpoint"]
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't create VPC endpoint for service %s. Here's why: %s: %s",
|
||
|
|
service_name,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ec2.CreateVpcEndpoint]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ec2.DeleteVpcEndpoints]
|
||
|
|
def delete_vpc_endpoints(self, vpc_endpoint_ids: list[str]) -> None:
|
||
|
|
"""
|
||
|
|
Deletes the specified VPC endpoints.
|
||
|
|
|
||
|
|
:param vpc_endpoint_ids: A list of IDs of the VPC endpoints to delete.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
self.ec2_client.delete_vpc_endpoints(VpcEndpointIds=vpc_endpoint_ids)
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't delete VPC endpoints %s. Here's why: %s: %s",
|
||
|
|
vpc_endpoint_ids,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ec2.DeleteVpcEndpoints]
|
||
|
|
|
||
|
|
# snippet-start:[python.example_code.ec2.DeleteVpc]
|
||
|
|
def delete(self, vpc_id: str) -> None:
|
||
|
|
"""
|
||
|
|
Deletes the specified VPC.
|
||
|
|
|
||
|
|
:param vpc_id: The ID of the VPC to delete.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
self.ec2_client.delete_vpc(VpcId=vpc_id)
|
||
|
|
except ClientError as err:
|
||
|
|
logger.error(
|
||
|
|
"Couldn't delete VPC %s. Here's why: %s: %s",
|
||
|
|
vpc_id,
|
||
|
|
err.response["Error"]["Code"],
|
||
|
|
err.response["Error"]["Message"],
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ec2.DeleteVpc]
|
||
|
|
|
||
|
|
|
||
|
|
# snippet-end:[python.example_code.ec2.VpcWrapper.class]
|