"""State module for managing Amazon API Rest APIs."""
import copy
from collections import OrderedDict
from dataclasses import field
from dataclasses import make_dataclass
from typing import Any
from typing import Dict
from typing import List
__contracts__ = ["resource"]
[docs]async def present(
hub,
ctx,
name: str,
resource_id: str = None,
description: str = None,
version: str = None,
binary_media_types: List[str] = None,
minimum_compression_size: int = None,
api_key_source: str = None,
endpoint_configuration: make_dataclass(
"EndpointConfiguration",
[
("types", List[str], field(default=None)),
("vpcEndpointIds", List[str], field(default=None)),
],
) = None,
policy: str = None,
disable_execute_api_endpoint: bool = None,
tags: Dict[str, Any] = None,
) -> Dict[str, Any]:
"""**Autogenerated function**
Creates a new rest_api resource or modifies an existing one.
Args:
name (str): An idem name of the resource.
resource_id (str, Optional): AWS rest_api id. Defaults to None.
description (str, Optional): The description of the RestApi.
version (str, Optional): A version identifier for the API.
binary_media_types (list[str], Optional): The list of binary media types supported by the RestApi.
By default, the RestApi supports only UTF-8-encoded text payloads.
minimum_compression_size (int, Optional): A nullable integer that is used to enable compression
(with non-negative between 0 and 10485760 (10M) bytes, inclusive) or disable compression
(with a null value) on an API. When compression is enabled, compression or decompression
is not applied on the payload if the payload size is smaller than this value. Setting it
to zero allows compression for any payload size.
api_key_source (str, Optional): The source of the API key for metering requests according to a usage plan.
endpoint_configuration (dict[str, Any], Optional):
The endpoint configuration of this RestApi showing the endpoint types of the API. Defaults to None.
* types (list[str], Optional):
A list of endpoint types of an API (RestApi) or its custom domain name (DomainName). For an
edge-optimized API and its custom domain name, the endpoint type is "EDGE". For a regional API
and its custom domain name, the endpoint type is REGIONAL. For a private API, the endpoint type
is PRIVATE.
* vpcEndpointIds (list[str], Optional):
A list of VpcEndpointIds of an API (RestApi) against which to create Route53 ALIASes. It is only
supported for PRIVATE endpoint type.
policy (str, Optional): A stringified JSON policy document that applies to this RestApi regardless of the caller and Method configuration.
tags (dict, Optional): The key-value map of strings. The valid character set is [a-zA-Z+-=._:/]. The tag key can be up
to 128 characters and must not start with aws:. The tag value can be up to 256 characters. Defaults to None.
disable_execute_api_endpoint (bool, Optional): Specifies whether clients can invoke your API by using the default execute-api endpoint. By default, clients can invoke your API with the default https://{api_id}.execute-api.{region}.amazonaws.com endpoint. To require that clients use a custom domain name to invoke your API, disable the default endpoint.
Returns:
dict[str, Any]
Examples:
.. code-block:: sls
idem_test_aws_apigateway_rest_api:
aws.apigateway.rest_api.present:
- name: test_rest_api
- endpoint_configuration:
types:
- PRIVATE
.. code-block:: sls
[idem_test_aws_apigateway_rest_api]:
aws.apigateway.rest_api.present:
- name: 'string'
- resource_id: 'string'
- api_key_source: 'string'
- binary_media_types:
- 'string'
- endpoint_configuration:
types:
- REGIONAL | EDGE | PRIVATE
vpcEndpointIds:
- 'string'
- tags:
'string': 'string'
- disable_execute_api_endpoint: True | False
- minimum_compression_size: int(0 to 10485760)
- policy: 'string'
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
before = None
resource_updated = False
if resource_id:
before = await hub.exec.aws.apigateway.rest_api.get(
ctx, name=name, resource_id=resource_id
)
if not before["result"]:
result["comment"] += before["comment"]
result["result"] = False
return result
result["old_state"] = before["ret"]
if before and before["ret"]:
result["new_state"] = copy.deepcopy(result["old_state"])
resource_arn = hub.tool.aws.arn_utils.build(
service="apigateway",
region=ctx["acct"]["region_name"],
resource="/restapis/" + resource_id,
)
updatable_resource_parameters = OrderedDict(
{
"description": description,
"binary_media_types": binary_media_types,
"minimum_compression_size": minimum_compression_size,
"api_key_source": api_key_source,
"endpoint_configuration": endpoint_configuration,
"disable_execute_api_endpoint": disable_execute_api_endpoint,
}
)
update_rest_api_ret = await hub.tool.aws.apigateway.rest_api.update_rest_api(
ctx,
old_state=result["old_state"],
updatable_resource_parameters=updatable_resource_parameters,
)
result["comment"] += update_rest_api_ret["comment"]
if not update_rest_api_ret["result"]:
result["result"] = False
return result
resource_updated = bool(update_rest_api_ret["ret"])
if resource_updated and ctx.get("test", False):
result["new_state"].update(update_rest_api_ret["ret"])
if tags is not None and tags != result["old_state"].get("tags", {}):
tag_update_ret = await hub.tool.aws.apigateway.tag.update(
ctx,
resource_arn=resource_arn,
old_tags=result["old_state"].get("tags", {}),
new_tags=tags,
)
result["comment"] += tag_update_ret["comment"]
if not tag_update_ret["result"]:
result["result"] = False
return result
resource_updated = resource_updated or bool(tag_update_ret["ret"])
if tag_update_ret["ret"] and ctx.get("test", False):
result["new_state"]["tags"] = tag_update_ret["ret"]
if resource_updated:
if ctx.get("test", False):
result["comment"] += hub.tool.aws.comment_utils.would_update_comment(
resource_type="aws.apigateway.rest_api", name=name
)
return result
else:
result["comment"] += hub.tool.aws.comment_utils.update_comment(
resource_type="aws.apigateway.rest_api", name=name
)
else:
result["comment"] += hub.tool.aws.comment_utils.already_exists_comment(
resource_type="aws.apigateway.rest_api", name=name
)
else:
if ctx.get("test", False):
result["new_state"] = hub.tool.aws.test_state_utils.generate_test_state(
enforced_state={},
desired_state={
"name": name,
"resource_id": resource_id,
"description": description,
"version": version,
"binary_media_types": binary_media_types,
"minimum_compression_size": minimum_compression_size,
"api_key_source": api_key_source,
"endpoint_configuration": endpoint_configuration,
"policy": policy,
"disable_execute_api_endpoint": disable_execute_api_endpoint,
"tags": tags,
},
)
result["comment"] += hub.tool.aws.comment_utils.would_create_comment(
resource_type="aws.apigateway.rest_api", name=name
)
return result
ret = await hub.exec.boto3.client.apigateway.create_rest_api(
ctx,
name=name,
description=description,
version=version,
binaryMediaTypes=binary_media_types,
minimumCompressionSize=minimum_compression_size,
apiKeySource=api_key_source,
endpointConfiguration=endpoint_configuration,
policy=policy,
tags=tags,
disableExecuteApiEndpoint=disable_execute_api_endpoint,
)
result["result"] = ret["result"]
if not result["result"]:
result["comment"] += ret["comment"]
return result
resource_id = ret["ret"]["id"]
result["comment"] += hub.tool.aws.comment_utils.create_comment(
resource_type="aws.apigateway.rest_api", name=name
)
if (not before) or resource_updated:
after = await hub.exec.aws.apigateway.rest_api.get(
ctx, name=name, resource_id=resource_id
)
if not (after["result"] and after["ret"]):
result["result"] = False
result["comment"] += after["comment"]
return result
result["new_state"] = after["ret"]
else:
result["new_state"] = copy.deepcopy(result["old_state"])
return result
[docs]async def absent(hub, ctx, name: str, resource_id: str = None) -> Dict[str, Any]:
"""**Autogenerated function**
Deletes the specified rest_api resource.
Args:
name (str): An Idem name of the resource.
resource_id (str): AWS rest_api id. Idem automatically considers this resource as absent if this field is not specified.
Returns:
dict[str, Any]
Examples:
.. code-block:: sls
resource_is_absent:
aws.apigateway.rest_api.absent:
- name: value
- resource_id: value
"""
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.apigateway.rest_api", name=name
)
return result
before = await hub.exec.aws.apigateway.rest_api.get(
ctx, name=name, resource_id=resource_id
)
if not (before["result"] and before["ret"]):
result["comment"] += hub.tool.aws.comment_utils.already_absent_comment(
resource_type="aws.apigateway.rest_api", name=name
)
if not ("NotFoundException" in str(before["comment"])):
result["comment"] += before["comment"]
result["result"] = False
return result
result["old_state"] = before["ret"]
if ctx.get("test", False):
result["comment"] += hub.tool.aws.comment_utils.would_delete_comment(
resource_type="aws.apigateway.rest_api", name=name
)
else:
ret = await hub.exec.boto3.client.apigateway.delete_rest_api(
ctx, restApiId=resource_id
)
result["result"] = ret["result"]
if not result["result"]:
result["comment"] += ret["comment"]
return result
result["comment"] += hub.tool.aws.comment_utils.delete_comment(
resource_type="aws.apigateway.rest_api", name=name
)
return result
[docs]async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]:
"""**Autogenerated function**
Describe the resource in a way that can be recreated/managed with the corresponding "present" function
Returns a list of apigateway.rest_api descriptions
Returns:
dict[str, Any]
Examples:
.. code-block:: bash
$ idem describe aws.apigateway.rest_api
"""
result = {}
ret = await hub.exec.boto3.client.apigateway.get_rest_apis(ctx)
if not ret["result"]:
hub.log.warning(f"Could not describe rest_api {ret['comment']}")
return {}
for rest_api in ret["ret"]["items"]:
full_rest_api = await hub.exec.aws.apigateway.rest_api.get(
ctx, name=rest_api["name"], resource_id=rest_api["id"]
)
if not full_rest_api["result"]:
hub.log.warning(
f"Could not describe rest_api '{rest_api['name']}': {ret['comment']}"
)
return {}
rest_api = full_rest_api["ret"]
name = rest_api.get("name")
# These 3 lines converting from NamespaceDict to dict are needed for rend <= 6.6.1;
# We're waiting for a new release which will remove this requirement.
rest_api = dict(rest_api)
rest_api["endpoint_configuration"] = dict(rest_api["endpoint_configuration"])
rest_api["tags"] = dict(rest_api["tags"])
result[name] = {
"aws.apigateway.rest_api.present": [{k: v} for k, v in rest_api.items()]
}
return result