"""States module for API Gateway v2 API resources."""
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"]
[docs]async def present(
hub,
ctx,
name: str,
protocol_type: str,
resource_id: str = None,
api_key_selection_expression: str = None,
cors_configuration: make_dataclass(
"Cors",
[
("AllowCredentials", bool, field(default=None)),
("AllowHeaders", List[str], field(default=None)),
("AllowMethods", List[str], field(default=None)),
("AllowOrigins", List[str], field(default=None)),
("ExposeHeaders", List[str], field(default=None)),
("MaxAge", int, field(default=None)),
],
) = None,
credentials_arn: str = None,
description: str = None,
disable_execute_api_endpoint: bool = None,
disable_schema_validation: bool = None,
route_key: str = None,
route_selection_expression: str = None,
tags: Dict[str, str] = None,
target: str = None,
version: str = None,
) -> Dict[str, Any]:
"""Creates an API Gateway v2 api resource.
Args:
name(str): An Idem name of the resource. This is also used as the name of the API during resource creation.
protocol_type(str): The API protocol.
resource_id(str, Optional): The API resource identifier in Amazon Web Services.
api_key_selection_expression(str, Optional): An API key selection expression. Supported only for WebSocket APIs.
cors_configuration(Dict[str, Any], Optional): A CORS configuration. Supported only for HTTP APIs. See Configuring CORS for more information. Defaults to None.
* AllowCredentials (bool, Optional): Specifies whether credentials are included in the CORS request. Supported only for HTTP APIs.
* AllowHeaders (List[str], Optional): Represents a collection of allowed headers. Supported only for HTTP APIs.
* AllowMethods (List[str], Optional): Represents a collection of allowed HTTP methods. Supported only for HTTP APIs.
* AllowOrigins (List[str], Optional): Represents a collection of allowed origins. Supported only for HTTP APIs.
* ExposeHeaders (List[str], Optional): Represents a collection of exposed headers. Supported only for HTTP APIs.
* MaxAge (int, Optional): The number of seconds that the browser should cache preflight request results. Supported only
for HTTP APIs.
credentials_arn(str, Optional): This property is part of quick create. It specifies the credentials required for the integration, if any.
For a Lambda integration, three options are available. To specify an IAM Role for API Gateway to assume, use the role's Amazon Resource Name (ARN).
To require that the caller's identity be passed through from the request, specify arn:aws:iam:::user/.
To use resource-based permissions on supported AWS services, specify null.
Currently, this property is not used for HTTP integrations. Supported only for HTTP APIs.
description(str, Optional): The description of the API.
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.
disable_schema_validation(bool, Optional): Avoid validating models when creating a deployment. Supported only for WebSocket APIs.
route_key(str, Optional): This property is part of quick create. If you don't specify a routeKey, a default route of $default is created.
The $default route acts as a catch-all for any request made to your API, for a particular stage.
The $default route key can't be modified. You can add routes after creating the API, and you can update the route keys of additional routes.
Supported only for HTTP APIs.
route_selection_expression(str, Optional): The route selection expression for the API. For HTTP APIs, the routeSelectionExpression must
be ${request.method} ${request.path}. If not provided, this will be the default for HTTP APIs. This property is required for WebSocket APIs.
tags(Dict, Optional): The collection of tags. Each tag element is associated with a given resource.
target(str, Optional): This property is part of quick create. Quick create produces an API with an integration, a default catch-all route,
and a default stage which is configured to automatically deploy changes. For HTTP integrations, specify a fully qualified URL.
For Lambda integrations, specify a function ARN. The type of the integration will be HTTP_PROXY or AWS_PROXY, respectively.
Supported only for HTTP APIs.
version(str, Optional): A version identifier for the API.
Request Syntax:
.. code-block:: sls
[idem_test_aws_apigatewayv2_api]:
aws.apigatewayv2.api.present:
- name: 'string'
- protocol_type: 'WEBSOCKET|HTTP'
- api_key_selection_expression: 'string'
- cors_configuration: Dict
- credentials_arn: 'string'
- description: 'string'
- disable_execute_api_endpoint: True|False
- disable_schema_validation: True|False
- route_key: 'string'
- route_selection_expression: 'string'
- tags:
- 'string': 'string'
- target: 'string'
- version: 'string'
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
idem_test_aws_apigatewayv2_api:
aws.apigatewayv2.api.present:
- name: value
- protocol_type: value
- api_key_selection_expression: value
- cors_configuration: { "AllowCredentials": True }
- credentials_arn: value
- description: value
- disable_execute_api_endpoint: True
- disable_schema_validation: True
- route_key: value
- route_selection_expression: value
- tags: { "key": "value" }
- target: value
- version: value
"""
result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
resource_updated = False
if resource_id:
before_ret = await hub.exec.aws.apigatewayv2.api.get(
ctx=ctx, name=name, resource_id=resource_id
)
if not before_ret["result"] or not before_ret["ret"]:
result["result"] = False
result["comment"] = before_ret["comment"]
return result
result["comment"] = hub.tool.aws.comment_utils.already_exists_comment(
resource_type="aws.apigatewayv2.api", name=name
)
result["old_state"] = copy.deepcopy(before_ret["ret"])
result["new_state"] = copy.deepcopy(result["old_state"])
resource_parameters = {
"api_key_selection_expression": api_key_selection_expression,
"cors_configuration": cors_configuration,
"credentials_arn": credentials_arn,
"description": description,
"disable_execute_api_endpoint": disable_execute_api_endpoint,
"disable_schema_validation": disable_schema_validation,
"route_key": route_key,
"route_selection_expression": route_selection_expression,
"target": target,
"version": version,
}
update_api_ret = await hub.tool.aws.apigatewayv2.api.update(
ctx,
resource_id=resource_id,
raw_resource=before_ret.get("ret"),
resource_parameters=resource_parameters,
)
result["comment"] += update_api_ret["comment"]
if not update_api_ret["result"]:
result["result"] = False
return result
resource_updated = resource_updated or bool(update_api_ret["ret"])
if update_api_ret["ret"] and ctx.get("test", False):
result["new_state"].update(update_api_ret["ret"])
if tags is not None and tags != result["old_state"].get("tags"):
resource_arn = hub.tool.aws.arn_utils.build(
service="apigateway",
region=ctx["acct"]["region_name"],
resource="/apis/" + resource_id,
)
update_tags_ret = await hub.tool.aws.apigatewayv2.tag.update_tags(
ctx,
resource_arn=resource_arn,
old_tags=result["old_state"].get("tags", {}),
new_tags=tags,
)
result["comment"] += update_tags_ret["comment"]
if not update_tags_ret["result"]:
result["result"] = False
return result
resource_updated = resource_updated or bool(update_tags_ret["ret"])
if update_tags_ret["ret"] and ctx.get("test", False):
result["new_state"]["tags"] = update_tags_ret["ret"].get("tags")
if resource_updated and ctx.get("test", False):
return result
else:
if ctx.get("test", False):
result[
"new_state"
] = raw_resource = hub.tool.aws.test_state_utils.generate_test_state(
enforced_state={},
desired_state={
"api_key_selection_expression": api_key_selection_expression,
"cors_configuration": cors_configuration,
"description": description,
"disable_execute_api_endpoint": disable_execute_api_endpoint,
"disable_schema_validation": disable_schema_validation,
"name": name,
"protocol_type": protocol_type,
"route_selection_expression": route_selection_expression,
"tags": tags,
"version": version,
},
)
result["comment"] = hub.tool.aws.comment_utils.would_create_comment(
resource_type="aws.apigatewayv2.api", name=name
)
return result
create_ret = await hub.exec.boto3.client.apigatewayv2.create_api(
ctx,
ApiKeySelectionExpression=api_key_selection_expression,
CorsConfiguration=cors_configuration,
CredentialsArn=credentials_arn,
Description=description,
DisableExecuteApiEndpoint=disable_execute_api_endpoint,
DisableSchemaValidation=disable_schema_validation,
Name=name,
ProtocolType=protocol_type,
RouteKey=route_key,
RouteSelectionExpression=route_selection_expression,
Tags=tags,
Target=target,
Version=version,
)
if not create_ret["result"]:
result["result"] = False
result["comment"] = create_ret["comment"]
return result
result["comment"] = hub.tool.aws.comment_utils.create_comment(
resource_type="aws.apigatewayv2.api", name=name
)
resource_id = create_ret["ret"]["ApiId"]
if (not result["old_state"]) or resource_updated:
after_ret = await hub.exec.aws.apigatewayv2.api.get(
ctx=ctx, name=name, resource_id=resource_id
)
if not after_ret["result"]:
result["result"] = False
result["comment"] += after_ret["comment"]
return result
result["new_state"] = after_ret["ret"]
return result
[docs]async def absent(hub, ctx, name: str, resource_id: str = None) -> Dict[str, Any]:
"""Deletes an API Gateway v2 api resource.
Args:
name(str): An Idem name of the resource.
resource_id(str, Optional): The API resource identifier in Amazon Web Services.
Idem automatically considers this resource being absent if this field is not specified.
Request syntax:
[idem_test_aws_apigatewayv2_api]:
aws.apigatewayv2.api.absent:
- name: 'string'
- resource_id: 'string'
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
idem_test_aws_apigatewayv2_api:
aws.apigatewayv2.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.apigatewayv2.api", name=name
)
return result
before_ret = await hub.exec.aws.apigatewayv2.api.get(
ctx=ctx, name=name, resource_id=resource_id
)
if not before_ret["result"]:
result["comment"] = before_ret["comment"]
result["result"] = False
return result
if not before_ret["ret"]:
result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
resource_type="aws.apigatewayv2.api", name=name
)
elif ctx.get("test", False):
result["old_state"] = before_ret["ret"]
result["comment"] = hub.tool.aws.comment_utils.would_delete_comment(
resource_type="aws.apigatewayv2.api", name=name
)
else:
result["old_state"] = before_ret["ret"]
delete_ret = await hub.exec.boto3.client.apigatewayv2.delete_api(
ctx, ApiId=resource_id
)
if not delete_ret["result"]:
result["result"] = False
result["comment"] = delete_ret["comment"]
return result
result["comment"] = hub.tool.aws.comment_utils.delete_comment(
resource_type="aws.apigatewayv2.api", name=name
)
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.
Gets the API Gateway v2 api resources for an AWS account.
Returns:
Dict[str, Dict[str, Any]]
Examples:
.. code-block:: bash
$ idem describe aws.apigatewayv2.api
"""
result = {}
describe_ret = await hub.exec.boto3.client.apigatewayv2.get_apis(ctx)
if not describe_ret["result"]:
hub.log.warning(f"Could not describe apis {describe_ret['comment']}")
return result
for api in describe_ret["ret"]["Items"]:
resource_translated = hub.tool.aws.apigatewayv2.api.convert_raw_api_to_present(
raw_resource=api,
)
result[resource_translated["name"]] = {
"aws.apigatewayv2.api.present": [
{parameter_key: parameter_value}
for parameter_key, parameter_value in resource_translated.items()
]
}
return result