"""
hub.exec.boto3.client.ec2.allocate_address
hub.exec.boto3.client.ec2.describe_addresses
hub.exec.boto3.client.ec2.release_address
"""
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 = {
"absent": {
"require": ["aws.ec2.nat_gateway.absent"],
}
}
[docs]async def present(
hub,
ctx,
name: str,
domain: str = "standard",
resource_id: str = None,
network_border_group: str = None,
public_ipv4_pool: str = None,
customer_owned_ipv4_pool: str = None,
tags: Dict[str, Any]
or List[
make_dataclass(
"Tag",
[("Key", str, field(default=None)), ("Value", str, field(default=None))],
)
] = None,
) -> Dict[str, Any]:
"""Allocates an Elastic IP address to your Amazon Web Services account.
After you allocate the Elastic IP address you can associate it with an instance or network interface. After you release an Elastic IP address,
it is released to the IP address pool and can be allocated to a different Amazon Web Services account. [EC2-VPC]
If you release an Elastic IP address, you might be able to recover it. You cannot recover an Elastic IP address
that you released after it is allocated to another Amazon Web Services account. You cannot recover an Elastic IP
address for EC2-Classic. To attempt to recover an Elastic IP address that you released, specify it in this
operation. An Elastic IP address is for use either in the EC2-Classic platform or in a VPC. By default,
you can allocate 5 Elastic IP addresses for EC2-Classic per Region and 5 Elastic IP addresses for EC2-VPC per
Region. For more information, see Elastic IP Addresses in the Amazon Elastic Compute Cloud User Guide.
Args:
name(str):
An Idem name of the elastic IP resource.
domain(str):
Indicates whether the Elastic IP address is for use with instances in a VPC or instances in EC2-Classic.
resource_id(str, Optional):
The public ip address.
network_border_group(str, Optional):
A unique set of Availability Zones, Local Zones, or Wavelength Zones from which Amazon Web Services
advertises IP addresses.
public_ipv4_pool(str, Optional):
The ID of an address pool that you own.
customer_owned_ipv4_pool(str, Optional):
The ID of a customer-owned address pool.
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 elastic IP resource. Defaults to None.
* Key (str):
The key of the tag. Tag keys are case-sensitive and accept a maximum of 127 Unicode characters. May not
begin with aws: .
* Value (str):
The value of the tag. Tag values are case-sensitive and accept a maximum of 255 Unicode characters.
Request Syntax:
.. code-block:: sls
[elastic_ip-resource-id]:
aws.ec2.elastic_ip.present:
- name: "string"
- domain: "string"
- tags:
scope: "string"
name: "string"
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
127.15.17.187:
aws.ec2.elastic_ip.present:
- name: 127.15.17.187
- resource_id: 127.15.17.187
- domain: vpc
- tags:
- Key: name
Value: ONE
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
allocation_id = None
public_ip = None
before = None
resource_updated = False
try:
if resource_id:
before = await hub.exec.aws.ec2.elastic_ip.get(
ctx, name=name, resource_id=resource_id
)
if isinstance(tags, List):
tags = hub.tool.aws.tag_utils.convert_tag_list_to_dict(tags)
if before:
if not before["result"] or not before["ret"]:
result["result"] = False
result["comment"] = before["comment"]
return result
public_ip = resource_id
result["old_state"] = copy.deepcopy(before["ret"])
plan_state = copy.deepcopy(result["old_state"])
result["comment"] += hub.tool.aws.comment_utils.already_exists_comment(
resource_type="aws.ec2.elastic_ip", name=resource_id
)
if domain == "vpc":
allocation_id = before["ret"]["allocation_id"]
old_tags = result["old_state"].get("tags")
if tags is not None and tags != old_tags:
update_ret = await hub.tool.aws.ec2.tag.update_tags(
ctx=ctx,
resource_id=allocation_id,
old_tags=old_tags,
new_tags=tags,
)
if not update_ret["result"]:
result["comment"] = update_ret["comment"]
result["result"] = False
return result
if update_ret["ret"]:
result["comment"] = update_ret["comment"]
resource_updated = resource_updated or bool(update_ret["result"])
if ctx.get("test", False) and update_ret["result"]:
plan_state["tags"] = update_ret["ret"]
else:
if ctx.get("test", False):
result["new_state"] = hub.tool.aws.test_state_utils.generate_test_state(
enforced_state={},
desired_state={
"name": name,
"domain": domain,
"network_border_group": network_border_group,
"public_ipv4_pool": public_ipv4_pool,
"customer_owned_ipv4_pool": customer_owned_ipv4_pool,
"allocation_id": "allocation_id_known_after_present",
"tags": tags,
},
)
result["comment"] = hub.tool.aws.comment_utils.would_create_comment(
resource_type="aws.ec2.elastic_ip", name=name
)
return result
ret = await hub.exec.boto3.client.ec2.allocate_address(
ctx,
Domain=domain,
PublicIpv4Pool=public_ipv4_pool,
NetworkBorderGroup=network_border_group,
CustomerOwnedIpv4Pool=customer_owned_ipv4_pool,
TagSpecifications=[
{
"ResourceType": "elastic-ip",
"Tags": hub.tool.aws.tag_utils.convert_tag_dict_to_list(tags),
}
]
if tags
else None,
)
result["result"] = ret["result"]
if not result["result"]:
result["result"] = ret["result"]
result["comment"] = ret["comment"]
return result
public_ip = ret["ret"]["PublicIp"]
result["comment"] = hub.tool.aws.comment_utils.create_comment(
resource_type="aws.ec2.elastic_ip", name=name
)
except Exception as e:
result["comment"] += [f"{e.__class__.__name__}: {e}"]
result["result"] = False
return result
try:
if ctx.get("test", False):
result["new_state"] = plan_state
elif (not before) or resource_updated:
after = await hub.exec.aws.ec2.elastic_ip.get(
ctx, name=name, resource_id=public_ip
)
result["new_state"] = copy.deepcopy(after["ret"])
else:
result["new_state"] = copy.deepcopy(result["old_state"])
except Exception as e:
result["comment"] += [f"{e.__class__.__name__}: {e}"]
result["result"] = False
return result
[docs]async def absent(hub, ctx, name: str, resource_id: str = None) -> Dict[str, Any]:
"""Releases the specified Elastic IP address.
[EC2-Classic, default VPC] Releasing an Elastic IP address automatically disassociates it from any instance that
it's associated with. To disassociate an Elastic IP address without releasing it, use DisassociateAddress .
[Nondefault VPC] You must use DisassociateAddress to disassociate the Elastic IP address before you can release it.
After releasing an Elastic IP address, it is released to the IP address pool. Be sure to update your DNS records and
any servers or devices that communicate with the address. If you attempt to release an Elastic IP address that you
already released, you'll get an AuthFailure error if the address is already allocated to another Amazon Web Services
account. [EC2-VPC] After you release an Elastic IP address for use in a VPC, you might be able to recover it. For
more information, see AllocateAddress.
Args:
name(str): An Idem name of the elastic IP resource.
resource_id(str, Optional): The public ip address. Idem automatically considers this resource being absent
if this field is not specified.
Request Syntax:
.. code-block:: sls
[elastic_ip-resource-id]:
aws.ec2.elastic_ip.absent:
- name: "string"
- resource_id: "string"
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
127.15.17.187:
aws.ec2.elastic_ip.absent:
- name: 127.15.17.187
- resource_id: 127.15.17.187
"""
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.elastic_ip", name=name
)
return result
before = await hub.exec.aws.ec2.elastic_ip.get(
ctx, name=name, resource_id=resource_id
)
if not before and not before["result"]:
result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
resource_type="aws.ec2.elastic_ip", name=name
)
if not before["ret"]:
result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
resource_type="aws.ec2.elastic_ip", name=name
)
elif ctx.get("test", False):
result["old_state"] = before["ret"]
result["comment"] = hub.tool.aws.comment_utils.would_delete_comment(
resource_type="aws.ec2.elastic_ip", name=name
)
return result
else:
result["old_state"] = before["ret"]
domain = result["old_state"]["domain"]
try:
if domain == "vpc":
allocation_id = result["old_state"]["allocation_id"]
ret = await hub.exec.boto3.client.ec2.release_address(
ctx, AllocationId=allocation_id
)
else:
ret = await hub.exec.boto3.client.ec2.release_address(
ctx, PublicIp=resource_id
)
result["result"] = ret["result"]
if not result["result"]:
result["comment"] = ret["comment"]
result["result"] = False
return result
result["comment"] = hub.tool.aws.comment_utils.delete_comment(
resource_type="aws.ec2.elastic_ip", name=name
)
except hub.tool.boto3.exception.ClientError as e:
result["comment"] += [f"{e.__class__.__name__}: {e}"]
result["result"] = False
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 the specified Elastic IP addresses or all of your Elastic IP addresses.
Returns:
Dict[str, Any]
Examples:
.. code-block:: bash
$ idem describe aws.ec2.elastic_ip
"""
result = {}
ret = await hub.exec.boto3.client.ec2.describe_addresses(ctx)
if not ret["result"]:
hub.log.warning(f"Could not describe elastic_ip {ret['comment']}")
return {}
for each_elastic_ip in ret["ret"]["Addresses"]:
resource_id = each_elastic_ip.get("PublicIp")
resource_translated = (
hub.tool.aws.ec2.conversion_utils.convert_raw_elastic_ip_to_present(
raw_resource=each_elastic_ip,
idem_resource_name=resource_id,
)
)
result[resource_id] = {
"aws.ec2.elastic_ip.present": [
{parameter_key: parameter_value}
for parameter_key, parameter_value in resource_translated.items()
]
}
return result