"""CFNgin blueprint base classes."""
from __future__ import annotations
import copy
import hashlib
import logging
import string
from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Dict,
List,
Optional,
Tuple,
Type,
TypeVar,
Union,
)
from troposphere import Output, Parameter, Ref, Template
from ...compat import cached_property
from ...mixins import DelCachedPropMixin
from ...variables import Variable
from ..exceptions import (
InvalidUserdataPlaceholder,
MissingVariable,
UnresolvedBlueprintVariable,
UnresolvedBlueprintVariables,
ValidatorError,
VariableTypeRequired,
)
from ..utils import read_value_from_path
from .variables.types import CFNType, TroposphereType
if TYPE_CHECKING:
from ...context import CfnginContext
from .type_defs import BlueprintVariableTypeDef
LOGGER = logging.getLogger(__name__)
PARAMETER_PROPERTIES = {
"default": "Default",
"description": "Description",
"no_echo": "NoEcho",
"allowed_values": "AllowedValues",
"allowed_pattern": "AllowedPattern",
"max_length": "MaxLength",
"min_length": "MinLength",
"max_value": "MaxValue",
"min_value": "MinValue",
"constraint_description": "ConstraintDescription",
}
_T = TypeVar("_T")
[docs]class CFNParameter:
"""Wrapper around a value to indicate a CloudFormation Parameter."""
[docs] def __init__(
self, name: str, value: Union[bool, float, int, List[Any], str, Any]
) -> None:
"""Instantiate class.
Args:
name: The name of the CloudFormation Parameter.
value: The value we're going to submit as a CloudFormation Parameter.
"""
self.name = name
if isinstance(value, (list, str)):
self.value = value
elif isinstance(value, bool):
LOGGER.debug("converting parameter %s boolean '%s' to string", name, value)
self.value = str(value).lower()
elif isinstance(value, (float, int)):
LOGGER.debug("converting parameter %s integer '%s' to string", name, value)
self.value = str(value)
else:
raise TypeError(
f"CFNParameter ({name}) value must be one of bool, float, int, str, "
f"List[str] but got: {type(value)}"
)
[docs] def __repr__(self) -> str:
"""Object represented as a string."""
return f"CFNParameter[{self.name}: {self.value}]"
@cached_property
def ref(self) -> Ref:
"""Ref the value of a parameter."""
return Ref(self.name)
[docs] def to_parameter_value(self) -> Union[List[Any], str]:
"""Return the value to be submitted to CloudFormation."""
return self.value
[docs]def build_parameter(name: str, properties: BlueprintVariableTypeDef) -> Parameter:
"""Build a troposphere Parameter with the given properties.
Args:
name: The name of the parameter.
properties: Contains the properties that will be applied to the parameter.
See: http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/parameters-section-structure.html
Returns:
The created parameter object.
""" # noqa: E501
param = Parameter(name, Type=properties.get("type"))
for name_, attr in PARAMETER_PROPERTIES.items():
if name_ in properties:
setattr(param, attr, properties[name_]) # type: ignore
return param
[docs]def validate_variable_type(
var_name: str,
var_type: Union[Type[CFNType], TroposphereType[Any], type],
value: Any,
) -> Any:
"""Ensure the value is the correct variable type.
Args:
var_name: The name of the defined variable on a blueprint.
var_type: The type that the value should be.
value: The object representing the value provided for the variable
Returns:
The appropriate value object. If the original value was of CFNType, the
returned value will be wrapped in CFNParameter.
Raises:
ValueError: If the `value` isn't of `var_type` and can't be cast as
that type, this is raised.
"""
if isinstance(var_type, TroposphereType):
try:
value = var_type.create(value)
except Exception as exc:
raise ValidatorError(
var_name, f"{var_type.resource_name}.create", value, exc
) from exc
elif issubclass(var_type, CFNType):
value = CFNParameter(name=var_name, value=value)
else:
if not isinstance(value, var_type):
raise TypeError(
f"Value for variable {var_name} must be of type {var_type}. Actual "
f"type: {type(value)}"
)
return value
[docs]def validate_allowed_values(allowed_values: Optional[List[Any]], value: Any) -> bool:
"""Support a variable defining which values it allows.
Args:
allowed_values: A list of allowed values from the variable definition.
value: The object representing the value provided for the variable.
Returns:
Boolean for whether or not the value is valid.
"""
# ignore CFNParameter, troposphere handles these for us
if not allowed_values or isinstance(value, CFNParameter):
return True
return value in allowed_values
[docs]def resolve_variable(
var_name: str,
var_def: BlueprintVariableTypeDef,
provided_variable: Optional[Variable],
blueprint_name: str,
) -> Any:
"""Resolve a provided variable value against the variable definition.
Args:
var_name: The name of the defined variable on a blueprint.
var_def: A dictionary representing the defined variables attributes.
provided_variable: The variable value provided to the blueprint.
blueprint_name: The name of the blueprint that the variable is being applied to.
Returns:
The resolved variable value, could be any python object.
Raises:
MissingVariable: Raised when a variable with no default is not
provided a value.
UnresolvedBlueprintVariable: Raised when the provided variable is
not already resolved.
ValueError: Raised when the value is not the right type and cannot be
cast as the correct type. Raised by
:func:`runway.cfngin.blueprints.base.validate_variable_type`
ValidatorError: Raised when a validator raises an exception. Wraps the
original exception.
"""
try:
var_type = var_def["type"]
except KeyError:
raise VariableTypeRequired(blueprint_name, var_name) from None
if provided_variable:
if not provided_variable.resolved:
raise UnresolvedBlueprintVariable(blueprint_name, provided_variable)
value = provided_variable.value
else:
# Variable value not provided, try using the default, if it exists
# in the definition
try:
value = var_def["default"]
except KeyError:
raise MissingVariable(blueprint_name, var_name) from None
# If no validator, return the value as is, otherwise apply validator
validator = var_def.get("validator", lambda v: v)
try:
value = validator(value)
except Exception as exc:
raise ValidatorError(var_name, validator.__name__, value, exc) from exc
# Ensure that the resulting value is the correct type
value = validate_variable_type(var_name, var_type, value)
allowed_values = var_def.get("allowed_values")
if not validate_allowed_values(allowed_values, value):
raise ValueError(
f"Invalid value passed to {var_name} in Blueprint {blueprint_name}. "
f"Got '{value}', expected one of {allowed_values}."
)
return value
[docs]def parse_user_data(
variables: Dict[str, Any], raw_user_data: str, blueprint_name: str
) -> str:
"""Parse the given user data and renders it as a template.
It supports referencing template variables to create userdata
that's supplemented with information from the stack, as commonly
required when creating EC2 userdata files.
Example:
Given a raw_user_data string: ``'open file ${file}'``
And a variables dictionary with: ``{'file': 'test.txt'}``
parse_user_data would output: ``open file test.txt``
Args:
variables: Variables available to the template.
raw_user_data: The user_data to be parsed.
blueprint_name: The name of the blueprint.
Returns:
The parsed user data, with all the variables values and refs replaced
with their resolved values.
Raises:
InvalidUserdataPlaceholder: Raised when a placeholder name in
raw_user_data is not valid. E.g ``${100}`` would raise this.
MissingVariable: Raised when a variable is in the raw_user_data that
is not given in the blueprint
"""
variable_values: Dict[str, Any] = {}
for key, value in variables.items():
if isinstance(value, CFNParameter):
variable_values[key] = value.to_parameter_value()
else:
variable_values[key] = value
template = string.Template(raw_user_data)
res = ""
try:
res = template.substitute(variable_values)
except ValueError as exc:
raise InvalidUserdataPlaceholder(blueprint_name, exc.args[0]) from exc
except KeyError as exc:
raise MissingVariable(blueprint_name, str(exc)) from exc
return res
[docs]class Blueprint(DelCachedPropMixin):
"""Base implementation for rendering a troposphere template.
Attributes:
VARIABLES: Class variable that defines the values that can be passed
from CFNgin to the template. These definition include metadata used
to validate the provided value and to propagate the variable to the
resulting CloudFormation template.
context: CFNgin context object.
description: The description of the CloudFormation template that will
be generated by this Blueprint.
mappings: CloudFormation Mappings to be added to the template during the
rendering process.
name: Name of the Stack that will be created by the Blueprint.
template: Troposphere template.
"""
VARIABLES: ClassVar[Dict[str, BlueprintVariableTypeDef]] = {}
context: CfnginContext
description: Optional[str]
mappings: Optional[Dict[str, Dict[str, Any]]]
name: str
template: Template
[docs] def __init__(
self,
name: str,
context: CfnginContext,
*,
description: Optional[str] = None,
mappings: Optional[Dict[str, Dict[str, Any]]] = None,
template: Optional[Template] = None,
**_: Any,
):
"""Instantiate class.
Args:
name: A name for the blueprint.
context: Context the blueprint is being executed under.
description: The description of the CloudFormation template that will
be generated by this blueprint.
mappings: CloudFormation Mappings to be used in the template during
the rendering process.
template: Optionally, provide a preexisting Template.
.. versionchanged:: 2.0.0
Added :attr:`template`.
.. versionchanged:: 2.0.0
Class only takes 2 positional arguments.
The rest are now keyword arguments.
"""
self._rendered = None
self._resolved_variables: Optional[Dict[str, Any]] = None
self._version = None
self.context = context
self.description = description
self.mappings = mappings
self.name = name
self.template = template or Template()
if hasattr(self, "PARAMETERS") or hasattr(self, "LOCAL_PARAMETERS"):
raise AttributeError(
f"DEPRECATION WARNING: Blueprint {name} uses "
"deprecated PARAMETERS or "
"LOCAL_PARAMETERS, rather than VARIABLES. "
"Please update your blueprints. See "
"https://docs.onica.com/projects/runway/page/cfngin/blueprints.html#variables "
"for additional information."
)
@cached_property
def cfn_parameters(self) -> Dict[str, Union[List[Any], str]]:
"""Return a dict of variables with type :class:`~runway.cfngin.blueprints.variables.types.CFNType`.
.. versionadded:: 2.0.0
Returns:
Variables that need to be submitted as CloudFormation Parameters.
""" # noqa
output: Dict[str, Union[List[Any], str]] = {}
for key, value in self.variables.items():
if hasattr(value, "to_parameter_value"):
output[key] = value.to_parameter_value()
return output
[docs] def create_template(self) -> None:
"""Abstract method called to create a template from the blueprint."""
raise NotImplementedError
@property
def defined_variables(self) -> Dict[str, BlueprintVariableTypeDef]:
"""Return a copy of :attr:`VARIABLES` to avoid accidental modification of the ClassVar.
.. versionchanged:: 2.0.0
Changed from a method to a property.
"""
return copy.deepcopy(self.VARIABLES)
@property
def output_definitions(self) -> Dict[str, Dict[str, Any]]:
"""Get the output definitions.
.. versionadded:: 2.0.0
Returns:
Output definitions. Keys are output names, the values are dicts
containing key/values for various output properties.
"""
return {k: output.to_dict() for k, output in self.template.outputs.items()}
@cached_property
def parameter_definitions(self) -> Dict[str, BlueprintVariableTypeDef]:
"""Get the parameter definitions to submit to CloudFormation.
Any variable definition whose type is an instance of
:class:`~runway.cfngin.blueprints.variables.types.CFNType` will
be returned as a CloudFormation Parameter.
.. versionadded:: 2.0.0
Returns:
Parameter definitions. Keys are parameter names, the values are dicts
containing key/values for various parameter properties.
"""
output: Dict[str, BlueprintVariableTypeDef] = {}
for var_name, attrs in self.defined_variables.items():
var_type = attrs.get("type")
if isinstance(var_type, type) and issubclass(var_type, CFNType):
cfn_attrs = copy.deepcopy(attrs)
cfn_attrs["type"] = var_type.parameter_type
output[var_name] = cfn_attrs
return output
@cached_property
def parameter_values(self) -> Dict[str, Union[List[Any], str]]:
"""Return a dict of variables with type :class:`~runway.cfngin.blueprints.variables.types.CFNType`.
.. versionadded:: 2.0.0
Returns:
Variables that need to be submitted as CloudFormation Parameters.
Will be a dictionary of <parameter name>: <parameter value>.
""" # noqa
output: Dict[str, Any] = {}
for key, value in self.variables.items():
try:
output[key] = value.to_parameter_value()
except AttributeError:
continue
return output
@property
def rendered(self) -> str:
"""Return rendered blueprint."""
if not self._rendered:
self._version, self._rendered = self.render_template()
return self._rendered
@cached_property
def required_parameter_definitions(self) -> Dict[str, BlueprintVariableTypeDef]:
"""Return all template parameters that do not have a default value.
.. versionadded:: 2.0.0
Returns:
Dict of required CloudFormation Parameters for the blueprint.
Will be a dictionary of ``<parameter name>: <parameter attributes>``.
"""
return {
name: attrs
for name, attrs in self.parameter_definitions.items()
if "Default" not in attrs
}
@property
def requires_change_set(self) -> bool:
"""Return true if the underlying template has transforms."""
return self.template.transform is not None
@property
def variables(self) -> Dict[str, Any]:
"""Return a Dict of variables available to the Template.
These variables will have been defined within :attr:`VARIABLES` or
:attr:`defined_variables`.
Any variable value that contains a Lookup will have been resolved.
.. versionadded:: 2.0.0
Returns:
Variables available to the Template.
Raises:
UnresolvedBlueprintVariables: If variables are unresolved.
"""
if self._resolved_variables is None:
raise UnresolvedBlueprintVariables(self.name)
return self._resolved_variables
@variables.setter
def variables(self, value: Dict[str, Any]) -> None:
"""Setter for :meth:`variables`.
.. versionadded:: 2.0.0
"""
self._resolved_variables = value
# clear cached properties that rely on this property
self._del_cached_property("cfn_parameters", "parameter_values")
@property
def version(self) -> str:
"""Template version."""
if not self._version:
self._version, self._rendered = self.render_template()
return self._version
[docs] def add_output(self, name: str, value: Any) -> None:
"""Add an output to the template.
Wrapper for ``self.template.add_output(Output(name, Value=value))``.
Args:
name: The name of the output to create.
value: The value to put in the output.
"""
self.template.add_output(Output(name, Value=value))
[docs] def get_cfn_parameters(self) -> Dict[str, Union[List[Any], str]]:
"""Return a dictionary of variables with `type` :class:`CFNType`.
.. deprecated:: 2.0.0
Replaced by :attr:`cfn_parameters`.
Returns:
Variables that need to be submitted as CloudFormation Parameters.
"""
LOGGER.warning(
"%s.get_cfn_parameters is deprecated and will be removed in a future release",
self.__class__.__name__,
)
return self.cfn_parameters
[docs] def get_output_definitions(self) -> Dict[str, Dict[str, Any]]:
"""Get the output definitions.
.. deprecated:: 2.0.0
Replaced by :attr:`output_definitions`.
Returns:
Output definitions. Keys are output names, the values are dicts
containing key/values for various output properties.
"""
LOGGER.warning(
"%s.get_output_definitions is deprecated and will be removed in a future release",
self.__class__.__name__,
)
return self.output_definitions
[docs] def get_parameter_definitions(self) -> Dict[str, BlueprintVariableTypeDef]:
"""Get the parameter definitions to submit to CloudFormation.
Any variable definition whose `type` is an instance of
:class:`~runway.cfngin.blueprints.variables.types.CFNType` will
be returned as a CloudFormation Parameter.
.. deprecated:: 2.0.0
Replaced by :attr:`parameter_definitions`.
Returns:
Parameter definitions. Keys are parameter names, the values are dicts
containing key/values for various parameter properties.
"""
LOGGER.warning(
"%s.get_parameter_definitions is deprecated and will be removed in a future release",
self.__class__.__name__,
)
return self.parameter_definitions
[docs] def get_parameter_values(self) -> Dict[str, Union[List[Any], str]]:
"""Return a dict of variables with type :class:`~runway.cfngin.blueprints.variables.types.CFNType`.
.. deprecated:: 2.0.0
Replaced by :attr:`parameter_values`.
Returns:
Variables that need to be submitted as CloudFormation Parameters.
Will be a dictionary of <parameter name>: <parameter value>.
""" # noqa
LOGGER.warning(
"%s.get_parameter_values is deprecated and will be removed in a future release",
self.__class__.__name__,
)
return self.parameter_values
[docs] def get_required_parameter_definitions(self) -> Dict[str, BlueprintVariableTypeDef]:
"""Return all template parameters that do not have a default value.
.. deprecated:: 2.0.0
Replaced by :attr:`required_parameter_definitions`.
Returns:
Dict of required CloudFormation Parameters for the blueprint.
Will be a dictionary of ``<parameter name>: <parameter attributes>``.
"""
LOGGER.warning(
"%s.get_required_parameter_definitions is deprecated and will be removed "
"in a future release",
self.__class__.__name__,
)
return self.required_parameter_definitions
[docs] def get_variables(self) -> Dict[str, Any]:
"""Return a dictionary of variables available to the template.
These variables will have been defined within `VARIABLES` or
`self.defined_variables`. Any variable value that contains a lookup
will have been resolved.
.. deprecated:: 2.0.0
Replaced by :attr:`variables`.
Returns:
Variables available to the template.
Raises:
UnresolvedBlueprintVariables: If variables are unresolved.
"""
LOGGER.warning(
"%s.get_variables is deprecated and will be removed in a future release",
self.__class__.__name__,
)
return self.variables
[docs] def import_mappings(self) -> None:
"""Import mappings from CFNgin config to the blueprint."""
if not self.mappings:
return
for name, mapping in self.mappings.items():
LOGGER.debug("adding mapping %s", name)
self.template.add_mapping(name, mapping)
[docs] def read_user_data(self, user_data_path: str) -> str:
"""Read and parse a user_data file.
Args:
user_data_path: Path to the userdata file.
"""
raw_user_data = read_value_from_path(user_data_path)
return parse_user_data(self.variables, raw_user_data, self.name)
[docs] def render_template(self) -> Tuple[str, str]:
"""Render the Blueprint to a CloudFormation template."""
self.import_mappings()
self.create_template()
if self.description:
self.set_template_description(self.description)
self.setup_parameters()
rendered = self.template.to_json(indent=self.context.template_indent)
version = hashlib.md5(rendered.encode()).hexdigest()[:8]
return version, rendered
[docs] def reset_template(self) -> None:
"""Reset template."""
self.template = Template()
self._rendered = None
self._version = None
[docs] def resolve_variables(self, provided_variables: List[Variable]) -> None:
"""Resolve the values of the blueprint variables.
This will resolve the values of the `VARIABLES` with values from the
env file, the config, and any lookups resolved.
Args:
provided_variables: List of provided variables.
"""
self._resolved_variables = {}
variable_dict = {var.name: var for var in provided_variables}
for var_name, var_def in self.defined_variables.items():
value = resolve_variable(
var_name, var_def, variable_dict.get(var_name), self.name
)
self._resolved_variables[var_name] = value
[docs] def set_template_description(self, description: str) -> None:
"""Add a description to the Template.
Args:
description: A description to be added to the resulting template.
"""
self.template.set_description(description)
[docs] def setup_parameters(self) -> None:
"""Add any CloudFormation parameters to the template."""
template = self.template
if not self.parameter_definitions:
LOGGER.debug("no parameters defined")
return
for name, attrs in self.parameter_definitions.items():
built_param = build_parameter(name, attrs)
template.add_parameter(built_param)
[docs] def to_json(self, variables: Optional[Dict[str, Any]] = None) -> str:
"""Render the blueprint and return the template in json form.
Args:
variables: Dictionary providing/overriding variable values.
"""
variables_to_resolve: List[Variable] = []
if variables:
for key, value in variables.items():
variables_to_resolve.append(Variable(key, value, "cfngin"))
for k in self.parameter_definitions:
if not variables or k not in variables:
# The provided value for a CFN parameter has no effect in this
# context (generating the CFN template), so any string can be
# provided for its value - just needs to be something
variables_to_resolve.append(Variable(k, "unused_value", "cfngin"))
self.resolve_variables(variables_to_resolve)
return self.render_template()[1]