"""
State module for managing EC2 Nat Gateways.
hub.exec.boto3.client.ec2.create_nat_gateway
hub.exec.boto3.client.ec2.delete_nat_gateway
hub.exec.boto3.client.ec2.describe_nat_gateways
"""
import copy
from dataclasses import field
from dataclasses import make_dataclass
from typing import Any
from typing import Dict
from typing import List
__contracts__ = ["resource"]
TREQ = {
"present": {
"require": ["aws.ec2.subnet.present"],
},
}
[docs]async def present(
hub,
ctx,
name: str,
subnet_id: str,
resource_id: str = None,
connectivity_type: str = "public",
client_token: str = None,
allocation_id: str = None,
tags: Dict[str, Any]
or List[
make_dataclass(
"Tag",
[("Key", str, field(default=None)), ("Value", str, field(default=None))],
)
] = None,
timeout: make_dataclass(
"Timeout",
[
(
"create",
make_dataclass(
"CreateTimeout",
[
("delay", int, field(default=None)),
("max_attempts", int, field(default=None)),
],
),
field(default=None),
),
(
"update",
make_dataclass(
"UpdateTimeout",
[
("delay", int, field(default=None)),
("max_attempts", int, field(default=None)),
],
),
field(default=None),
),
],
) = None,
) -> Dict[str, Any]:
"""Creates a NAT gateway in the specified subnet.
This action creates a network interface in the specified subnet with a private IP address from the IP address range
of the subnet. You can create either a public NAT gateway or a private NAT gateway. With a public NAT gateway,
internet-bound traffic from a private subnet can be routed to the NAT gateway, so that instances in a private subnet
can connect to the internet. With a private NAT gateway, private communication is routed across VPCs and on-premises
networks through a transit gateway or virtual private gateway. Common use cases include running large workloads behind
a small pool of allowlisted IPv4 addresses, preserving private IPv4 addresses, and communicating between overlapping
networks. For more information, see NAT gateways in the Amazon Virtual Private Cloud User Guide.
Args:
name(str):
An Idem name to identify the NAT gateway resource.
subnet_id(str):
The subnet in which to create the NAT gateway.
resource_id(str, Optional):
AWS Internet Gateway ID.
client_token(str, Optional):
Unique, case-sensitive identifier that you provide to ensure the idempotency of the request. Constraint: Maximum 64 ASCII characters. This field is autopopulated if not provided.
connectivity_type(str, Optional):
Indicates whether the NAT gateway supports public or private connectivity. The default is public connectivity.
allocation_id(str, Optional):
[Public NAT gateway only]. The allocation ID of the Elastic IP address that's associated with the NAT gateway.
tags(dict or list, Optional):
Dict in the format of ``{tag-key: tag-value}`` or List of tags in the format of
``[{"Key": tag-key, "Value": tag-value}]`` to associate with the NAT gateway resource. Defaults to None.
* (Key, Optional):
The key of the tag. Constraints: Tag keys are case-sensitive and accept a maximum of 127 Unicode
characters. May not begin with aws:.
* (Value, Optional):
The value of the tag. Constraints: Tag values are case-sensitive and accept a maximum of 256
Unicode characters.
timeout(dict, Optional):
Timeout configuration for create/update/deletion of AWS IAM Policy.
* create (dict):
Timeout configuration for creating AWS IAM Policy
* delay (int, Optional): The amount of time in seconds to wait between attempts.
* max_attempts (int, Optional): Customized timeout configuration containing delay and max attempts.
* update(dict, Optional):
Timeout configuration for updating AWS IAM Policy
* delay (int, Optional):
The amount of time in seconds to wait between attempts.
* max_attempts (int, Optional):
Customized timeout configuration containing delay and max attempts.
Request Syntax:
.. code-block:: sls
[nat_gateway-name]:
aws.ec2.nat_gateway.present:
- subnet_id: 'string'
- resource_id: 'string'
- client_token: 'string'
- connectivity_type: 'string'
- allocation_id: 'string'
- tags:
- Key: 'string'
Value: 'string'
- timeout:
create:
delay: 'integer'
max_attempts: 'integer'
update:
delay: 'integer'
max_attempts: 'integer'
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
my-nat-gateway:
aws.ec2.nat_gateway.present:
- subnet_id: subnet-026542dd856a97e15
- client_token: nat-0e7969ddf4c38831f
- resource_id: nat-0e7969ddf4c38831f
- connectivity_type: private
- tags:
- Key: Name
Value: test-nat-gateway
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
before = None
resource_updated = False
if isinstance(tags, List):
tags = hub.tool.aws.tag_utils.convert_tag_list_to_dict(tags)
if resource_id:
before = await hub.exec.boto3.client.ec2.describe_nat_gateways(
ctx, NatGatewayIds=[resource_id]
)
if before and before["result"] and before["ret"].get("NatGateways"):
result[
"old_state"
] = hub.tool.aws.ec2.conversion_utils.convert_raw_nat_gateway_to_present(
raw_resource=before["ret"]["NatGateways"][0], idem_resource_name=name
)
old_tags = result["old_state"].get("tags")
plan_state = copy.deepcopy(result["old_state"])
result["comment"] += hub.tool.aws.comment_utils.already_exists_comment(
resource_type="aws.ec2.nat_gateway", name=resource_id
)
# check if tags need to be updated
if tags is not None and tags != old_tags:
update_ret = await hub.tool.aws.ec2.tag.update_tags(
ctx=ctx,
resource_id=resource_id,
old_tags=old_tags,
new_tags=tags,
)
if not update_ret["result"]:
result["comment"] += update_ret["comment"]
result["result"] = False
elif ctx.get("test", False):
plan_state["tags"] = update_ret["ret"]
result["comment"] += [
f"Would update tags for aws.ec2.nat_gateway {name}"
]
else:
resource_updated = True
result["comment"] += update_ret["comment"]
if resource_updated:
waiter_config = hub.tool.aws.waiter_utils.create_waiter_config(
default_delay=15,
default_max_attempts=40,
timeout_config=timeout.get("update") if timeout else None,
)
try:
await hub.tool.boto3.client.wait(
ctx,
"ec2",
"nat_gateway_available",
NatGatewayIds=[resource_id],
WaiterConfig=waiter_config,
)
except Exception as e:
result["comment"] += [str(e)]
result["result"] = False
elif not ctx.get("test", False):
result["comment"] += [
f"aws.ec2.nat_gateway '{name}' has no property need to be updated"
]
else:
if ctx.get("test", False):
result["new_state"] = hub.tool.aws.test_state_utils.generate_test_state(
enforced_state={},
desired_state={
"allocation_id": allocation_id,
"client_token": client_token,
"subnet_id": subnet_id,
"name": name,
"tags": tags,
"connectivity_type": connectivity_type,
},
)
result["comment"] += [f"Would create aws.ec2.nat_gateway '{name}'"]
return result
else:
ret = await hub.exec.boto3.client.ec2.create_nat_gateway(
ctx,
AllocationId=allocation_id,
ClientToken=client_token,
SubnetId=subnet_id,
TagSpecifications=[
{
"ResourceType": "natgateway",
"Tags": hub.tool.aws.tag_utils.convert_tag_dict_to_list(tags),
}
]
if tags is not None
else None,
ConnectivityType=connectivity_type,
)
result["result"] = ret["result"]
if not result["result"]:
result["comment"] = ret["comment"]
return result
resource_id = ret["ret"]["NatGateway"]["NatGatewayId"]
waiter_config = hub.tool.aws.waiter_utils.create_waiter_config(
default_delay=15,
default_max_attempts=40,
timeout_config=timeout.get("create") if timeout else None,
)
try:
await hub.tool.boto3.client.wait(
ctx,
"ec2",
"nat_gateway_available",
NatGatewayIds=[resource_id],
WaiterConfig=waiter_config,
)
except Exception as e:
result["comment"] += [str(e)]
result["result"] = False
if result["result"]:
result["comment"] += [f"Created '{resource_id}'"]
try:
if ctx.get("test", False):
result["new_state"] = plan_state
elif (not before) or resource_updated:
after = await hub.exec.boto3.client.ec2.describe_nat_gateways(
ctx, NatGatewayIds=[resource_id]
)
result[
"new_state"
] = hub.tool.aws.ec2.conversion_utils.convert_raw_nat_gateway_to_present(
raw_resource=after["ret"]["NatGateways"][0], idem_resource_name=name
)
else:
result["new_state"] = copy.deepcopy(result["old_state"])
except Exception as e:
result["comment"] += [str(e)]
result["result"] = False
return result
[docs]async def absent(
hub, ctx, name: str, resource_id: str = None, timeout: Dict = None
) -> Dict[str, Any]:
"""Deletes the specified NAT gateway.
Deleting a public NAT gateway disassociates its Elastic IP address, but does not release the address from your
account. Deleting a NAT gateway does not delete any NAT gateway routes in your route tables.
Args:
name(str):
An Idem name to identify the NAT gateway resource.
resource_id(str, Optional):
The AWS ID of the nat gateway. Idem automatically considers this resource being absent
if this field is not specified.
timeout(dict, Optional):
Timeout configuration for deletion of AWS Nat Gateway.
* delete (dict):
Timeout configuration for deletion of a Nat Gateway
* delay: The amount of time in seconds to wait between attempts.
* max_attempts: Customized timeout configuration containing delay and max attempts.
Request Syntax:
.. code-block:: sls
[nat_gateway-name]:
aws.ec2.nat_gateway.absent:
- name: 'string'
- resource_id: 'string'
- timeout:
delete:
delay: 'integer'
max_attempts: 'integer'
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
my-nat-gateway:
aws.ec2.nat_gateway.absent:
- name: my-nat-gateway
- resource_id: nat-0e7969ddf4c38831f
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
if not resource_id:
result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
resource_type="aws.ec2.nat_gateway", name=name
)
return result
before = await hub.exec.boto3.client.ec2.describe_nat_gateways(
ctx, NatGatewayIds=[resource_id]
)
if not before["result"]:
result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
resource_type="aws.ec2.nat_gateway", name=name
)
elif ctx.get("test", False):
result[
"old_state"
] = hub.tool.aws.ec2.conversion_utils.convert_raw_nat_gateway_to_present(
raw_resource=before, idem_resource_name=name
)
result["comment"] += [f"Would delete aws.ec2.nat_gateway {name}"]
return result
else:
try:
if before["ret"]["NatGateways"][0]["State"] == "deleted":
result["comment"] += [
f"aws.ec2.nat_gateway '{name}' is in deleted state."
]
else:
result[
"old_state"
] = hub.tool.aws.ec2.conversion_utils.convert_raw_nat_gateway_to_present(
raw_resource=before["ret"]["NatGateways"][0],
idem_resource_name=name,
)
ret = await hub.exec.boto3.client.ec2.delete_nat_gateway(
ctx, NatGatewayId=resource_id
)
result["result"] = ret["result"]
if not result["result"]:
result["comment"] = ret["comment"]
return result
waiter_config = hub.tool.aws.waiter_utils.create_waiter_config(
default_delay=15,
default_max_attempts=40,
timeout_config=timeout.get("delete") if timeout else None,
)
acceptors = [
{
"matcher": "pathAll",
"expected": "deleted",
"state": "success",
"argument": "NatGateways[].State",
},
{
"matcher": "pathAll",
"expected": "deleting",
"state": "retry",
"argument": "NatGateways[].State",
},
]
cluster_waiter = hub.tool.boto3.custom_waiter.waiter_wrapper(
name="NatGatewayDeleted",
operation="DescribeNatGateways",
argument=["NatGateways[].State"],
acceptors=acceptors,
client=await hub.tool.boto3.client.get_client(ctx, "ec2"),
)
await hub.tool.boto3.client.wait(
ctx,
"ec2",
"NatGatewayDeleted",
cluster_waiter,
NatGatewayIds=[resource_id],
WaiterConfig=waiter_config,
)
result["comment"] += [f"Deleted '{name}'"]
except hub.tool.boto3.exception.ClientError as e:
result["comment"] += [f"{e.__class__.__name__}: {e}"]
return result
[docs]async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]:
"""
Describe the resource in a way that can be recreated/managed with the corresponding "present" function
Describes one or more of your NAT gateways.
Returns:
Dict[str, Any]
Examples:
.. code-block:: bash
$ idem describe aws.ec2.nat_gateway
"""
result = {}
ret = await hub.exec.boto3.client.ec2.describe_nat_gateways(ctx)
if not ret["result"]:
hub.log.warning(f"Could not describe nat_gateway {ret['comment']}")
return result
for nat_gateway in ret["ret"]["NatGateways"]:
nat_gateway_id = nat_gateway.get("NatGatewayId")
resource_translated = (
hub.tool.aws.ec2.conversion_utils.convert_raw_nat_gateway_to_present(
raw_resource=nat_gateway, idem_resource_name=nat_gateway_id
)
)
result[nat_gateway_id] = {
"aws.ec2.nat_gateway.present": [
{parameter_key: parameter_value}
for parameter_key, parameter_value in resource_translated.items()
]
}
return result