# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2022 John Mille <john@compose-x.io>
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from troposphere import AWSObject
from .settings import ComposeXSettings
from .stacks import ComposeXStack
from copy import deepcopy
from compose_x_common.compose_x_common import keyisset, keypresent, set_else_none
from troposphere import AWS_NO_VALUE, Join, Output
from troposphere import Parameter as CfnParameter
from troposphere import Ref, Template
from ecs_composex import __version__ as version
from ecs_composex.common import DATE, cfn_conditions
from ecs_composex.common.cfn_params import ROOT_STACK_NAME, Parameter
[docs]def no_value_if_not_set(props, key, is_bool=False):
"""
Function to simplify setting value if the key is in the dict and else Ref(AWS_NO_VALUE) for resource properties
:param dict props:
:param str key:
:param bool is_bool:
:return:
"""
if not is_bool:
return Ref(AWS_NO_VALUE) if not keyisset(key, props) else props[key]
else:
return Ref(AWS_NO_VALUE) if not keypresent(key, props) else props[key]
[docs]def init_template(description=None):
"""Function to initialize the troposphere base template
:param description: Description used for the CFN
:type description: str
:returns: template
:rtype: Template
"""
if description is not None:
template = Template(description)
else:
template = Template("Template generated by ECS ComposeX")
template.set_metadata(
deepcopy(
{
"Type": "ComposeX",
"Properties": {"Version": version, "GeneratedOn": DATE},
}
)
)
template.set_version()
return template
[docs]def add_parameter_to_group_label(
interface_metadata: dict, parameter: Parameter
) -> None:
"""
Simply goes over the ParameterGroups of the metadata.AWS::CloudFormation::Interface
and if already exists, adds to group, else, create group and adds first element
:param dict interface_metadata:
:param ecs_composex.common.cfn_params.Parameter parameter:
"""
groups = set_else_none("ParameterGroups", interface_metadata, [], eval_bool=True)
if not groups:
interface_metadata["ParameterGroups"] = groups
groups.append(
{
"Label": {"default": parameter.group_label},
"Parameters": [parameter.title],
}
)
else:
for group in groups:
if group["Label"]["default"] == parameter.group_label:
if parameter.title not in group["Parameters"]:
group["Parameters"].append(parameter.title)
break
else:
groups.append(
{
"Label": {"default": parameter.group_label},
"Parameters": [parameter.title],
}
)
[docs]def add_parameters(template: Template, parameters: list) -> None:
"""Function to add parameters to the template
:param template: the template to add the parameters to
:type template: troposphere.Template
:param parameters: list of parameters to add to the template
:type parameters: list<ecs_composex.common.cfn_params.Parameter>
"""
for param in parameters:
if not isinstance(param, (Parameter, CfnParameter)) or not issubclass(
type(param), (Parameter, CfnParameter)
):
raise TypeError("Parameter must be of type", Parameter, "Got", type(param))
if template and param.title not in template.parameters:
template.add_parameter(param)
if isinstance(param, Parameter) and (param.group_label or param.label):
add_parameters_metadata(template, param)
[docs]def add_outputs(template, outputs):
"""Function to add parameters to the template
:param template: the template to add the parameters to
:type template: troposphere.Template
:param outputs: list of parameters to add to the template
:type outputs: list<troposphere.Output>
"""
for output in outputs:
if not isinstance(output, Output):
raise TypeError("Parameter must be of type", Output)
if template and output.title not in template.outputs:
template.add_output(output)
[docs]def add_update_mapping(template, mapping_key, mapping_value, mapping_subkey=None):
"""
:param troposphere.Template template:
:param str mapping_key:
:param dict mapping_value:
:param str mapping_subkey: If set, applies the value to a sub-key of the mapping on update
:return:
"""
if mapping_key not in template.mappings:
template.add_mapping(mapping_key, mapping_value)
else:
if mapping_subkey and keyisset(mapping_subkey, template.mappings[mapping_key]):
template.mappings[mapping_key][mapping_subkey].update(mapping_value)
elif (
mapping_key
and mapping_subkey
and not keyisset(mapping_subkey, template.mappings[mapping_key])
):
template.mappings[mapping_key][mapping_subkey] = mapping_value
elif not mapping_subkey:
template.mappings[mapping_key].update(mapping_value)
[docs]def add_resource(template, resource, replace=False) -> AWSObject:
"""
Function to add resource to template if the resource does not already exist
:param troposphere.Template template:
:param troposphere.AWSObject resource:
:param bool replace:
"""
if (
resource not in template.resources.values()
and resource.title not in template.resources.keys()
):
return template.add_resource(resource)
elif resource.title in template.resources and replace:
template.resources[resource.title] = resource
return resource
[docs]def set_get_resource(template, resource) -> (AWSObject, bool):
"""
Function to add resource to template if the resource does not already exist
Returns the resource if it already does.
"""
if (
resource not in template.resources.values()
and resource.title not in template.resources.keys()
):
return template.add_resource(resource), False
return template.resources[resource.title], True
[docs]def add_defaults(template):
"""Function to CFN parameters and conditions to the template which are used
across ECS ComposeX
:param template: source template to add the params and conditions to
:type template: Template
"""
template.add_parameter(ROOT_STACK_NAME)
template.add_condition(
cfn_conditions.USE_STACK_NAME_CON_T, cfn_conditions.USE_STACK_NAME_CON
)
[docs]def build_template(description=None, *parameters):
"""
Entry point function to creating the template for ECS ComposeX resources
:param description: Optional custom description for the CFN template
:type description: str, optional
:param parameters: List of optional parameters to add to the template.
:type parameters: List<troposphere.Parameters>, optional
:returns template: the troposphere template
:rtype: Template
"""
template = init_template(description)
if parameters:
add_parameters(template, *parameters)
add_defaults(template)
return template
[docs]def add_update_parameter_recursively(
ext_stack: ComposeXStack,
compose_settings: ComposeXSettings,
attribute_settings: dict,
):
"""
Recursively adds parameters to an external stack and updates
the parameters as it goes.
if current external stack has no parent or the parent is the root stack,
use
`attribute_settings["ImportValue"]` is the `GetAtt stack.Outputs.<output_name>` for that value.
Otherwise, we consider that we have started from a lower node, we add the direct reference to the value,
and recursively go up the parent stacks until the first condition is met.
We set the value to `Ref` given the value is given via parameter coming from the parent stack.
If however the parameter is a List<> as defined in the AWS CFN Docs, we flatten the list with Join for nested
Stack parameters.
"""
attribute_param = attribute_settings["ImportParameter"]
if (
ext_stack.parent_stack
and (ext_stack.parent_stack == compose_settings.root_stack)
) or not ext_stack.parent_stack:
add_parameters(ext_stack.stack_template, [attribute_param])
ext_stack.Parameters.update(
{attribute_param.title: attribute_settings["ImportValue"]}
)
else:
add_parameters(ext_stack.stack_template, [attribute_param])
if isinstance(attribute_param, Parameter) and attribute_param.Type.startswith(
r"List<"
):
value = Join(",", Ref(attribute_param))
else:
value = Ref(attribute_param)
ext_stack.Parameters.update({attribute_param.title: value})
return add_update_parameter_recursively(
ext_stack.parent_stack, compose_settings, attribute_settings
)
return Ref(attribute_param)