Source code for infrahouse_core.aws.vpc_endpoint
"""
VPC Endpoint resource wrapper.
Provides ``exists`` / ``delete()`` support for EC2 VPC endpoints.
"""
from __future__ import annotations
from logging import getLogger
from botocore.exceptions import ClientError
from infrahouse_core.aws.base import AWSResource
LOG = getLogger(__name__)
[docs]
class VPCEndpoint(AWSResource):
"""Wrapper around an EC2 VPC Endpoint.
:param vpc_endpoint_id: VPC endpoint ID (e.g. ``vpce-0123456789abcdef0``).
:param region: AWS region.
:param role_arn: IAM role ARN for cross-account access.
"""
def __init__(self, vpc_endpoint_id, region=None, role_arn=None, session=None):
super().__init__(vpc_endpoint_id, "ec2", region=region, role_arn=role_arn, session=session)
@property
def vpc_endpoint_id(self) -> str:
"""Return the VPC endpoint ID.
:rtype: str
"""
return self._resource_id
@property
def exists(self) -> bool:
"""Return ``True`` if the VPC endpoint exists and is not deleted."""
try:
response = self._client.describe_vpc_endpoints(VpcEndpointIds=[self._resource_id])
endpoints = response.get("VpcEndpoints", [])
if not endpoints:
return False
return endpoints[0].get("State") != "deleted"
except ClientError as err:
if err.response["Error"]["Code"] == "InvalidVpcEndpointId.NotFound":
return False
raise
[docs]
def delete(self) -> None:
"""Delete the VPC endpoint.
Idempotent -- does nothing if the endpoint does not exist.
"""
try:
self._client.delete_vpc_endpoints(VpcEndpointIds=[self._resource_id])
LOG.info("Deleted VPC endpoint %s", self._resource_id)
except ClientError as err:
if err.response["Error"]["Code"] == "InvalidVpcEndpointId.NotFound":
LOG.info("VPC endpoint %s does not exist.", self._resource_id)
else:
raise