Source code for heat.engine.cfn.template

#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import functools

from heat.common import exception
from heat.common.i18n import _
from heat.engine.cfn import functions as cfn_funcs
from heat.engine.cfn import parameters as cfn_params
from heat.engine import function
from heat.engine import parameters
from heat.engine import rsrc_defn
from heat.engine import template_common


[docs]class CfnTemplateBase(template_common.CommonTemplate): """The base implementation of cfn template.""" SECTIONS = ( VERSION, ALTERNATE_VERSION, DESCRIPTION, MAPPINGS, PARAMETERS, RESOURCES, OUTPUTS, ) = ( 'AWSTemplateFormatVersion', 'HeatTemplateFormatVersion', 'Description', 'Mappings', 'Parameters', 'Resources', 'Outputs', ) OUTPUT_KEYS = ( OUTPUT_DESCRIPTION, OUTPUT_VALUE, ) = ( 'Description', 'Value', ) SECTIONS_NO_DIRECT_ACCESS = set([PARAMETERS, VERSION, ALTERNATE_VERSION]) _RESOURCE_KEYS = ( RES_TYPE, RES_PROPERTIES, RES_METADATA, RES_DEPENDS_ON, RES_DELETION_POLICY, RES_UPDATE_POLICY, RES_DESCRIPTION, ) = ( 'Type', 'Properties', 'Metadata', 'DependsOn', 'DeletionPolicy', 'UpdatePolicy', 'Description', ) functions = { 'Fn::FindInMap': cfn_funcs.FindInMap, 'Fn::GetAZs': cfn_funcs.GetAZs, 'Ref': cfn_funcs.Ref, 'Fn::GetAtt': cfn_funcs.GetAtt, 'Fn::Select': cfn_funcs.Select, 'Fn::Join': cfn_funcs.Join, 'Fn::Base64': cfn_funcs.Base64, } deletion_policies = { 'Delete': rsrc_defn.ResourceDefinition.DELETE, 'Retain': rsrc_defn.ResourceDefinition.RETAIN, 'Snapshot': rsrc_defn.ResourceDefinition.SNAPSHOT } HOT_TO_CFN_RES_ATTRS = {'type': RES_TYPE, 'properties': RES_PROPERTIES, 'metadata': RES_METADATA, 'depends_on': RES_DEPENDS_ON, 'deletion_policy': RES_DELETION_POLICY, 'update_policy': RES_UPDATE_POLICY} HOT_TO_CFN_OUTPUT_ATTRS = {'description': OUTPUT_DESCRIPTION, 'value': OUTPUT_VALUE} def __getitem__(self, section): """Get the relevant section in the template.""" if section not in self.SECTIONS: raise KeyError(_('"%s" is not a valid template section') % section) if section in self.SECTIONS_NO_DIRECT_ACCESS: raise KeyError( _('Section %s can not be accessed directly.') % section) if section == self.DESCRIPTION: default = 'No description' else: default = {} # if a section is None (empty yaml section) return {} # to be consistent with an empty json section. return self.t.get(section) or default
[docs] def param_schemata(self, param_defaults=None): params = self.t.get(self.PARAMETERS) or {} pdefaults = param_defaults or {} for name, schema in params.items(): if name in pdefaults: params[name][parameters.DEFAULT] = pdefaults[name] return dict((name, parameters.Schema.from_dict(name, schema)) for name, schema in params.items())
[docs] def get_section_name(self, section): return section
[docs] def parameters(self, stack_identifier, user_params, param_defaults=None): return cfn_params.CfnParameters(stack_identifier, self, user_params=user_params, param_defaults=param_defaults)
[docs] def resource_definitions(self, stack): resources = self.t.get(self.RESOURCES) or {} conditions = self.conditions(stack) def defns(): for name, snippet in resources.items(): try: defn_data = dict(self._rsrc_defn_args(stack, name, snippet)) except (TypeError, ValueError, KeyError) as ex: msg = str(ex) raise exception.StackValidationFailed(message=msg) defn = rsrc_defn.ResourceDefinition(name, **defn_data) cond_name = defn.condition() if cond_name is not None: try: enabled = conditions.is_enabled(cond_name) except ValueError as exc: path = [self.RESOURCES, name, self.RES_CONDITION] message = str(exc) raise exception.StackValidationFailed(path=path, message=message) if not enabled: continue yield name, defn return dict(defns())
[docs] def add_resource(self, definition, name=None): if name is None: name = definition.name hot_tmpl = definition.render_hot() if self.t.get(self.RESOURCES) is None: self.t[self.RESOURCES] = {} cfn_tmpl = dict((self.HOT_TO_CFN_RES_ATTRS[k], v) for k, v in hot_tmpl.items()) dep_list = cfn_tmpl.get(self.RES_DEPENDS_ON, []) if len(dep_list) == 1: dep_res = cfn_tmpl[self.RES_DEPENDS_ON][0] if dep_res in self.t[self.RESOURCES]: cfn_tmpl[self.RES_DEPENDS_ON] = dep_res else: del cfn_tmpl[self.RES_DEPENDS_ON] elif dep_list: cfn_tmpl[self.RES_DEPENDS_ON] = [d for d in dep_list if d in self.t[self.RESOURCES]] self.t[self.RESOURCES][name] = cfn_tmpl
[docs] def add_output(self, definition): hot_op = definition.render_hot() cfn_op = dict((self.HOT_TO_CFN_OUTPUT_ATTRS[k], v) for k, v in hot_op.items()) if self.t.get(self.OUTPUTS) is None: self.t[self.OUTPUTS] = {} self.t[self.OUTPUTS][definition.name] = cfn_op
[docs]class CfnTemplate(CfnTemplateBase): CONDITIONS = 'Conditions' SECTIONS = CfnTemplateBase.SECTIONS + (CONDITIONS,) SECTIONS_NO_DIRECT_ACCESS = (CfnTemplateBase.SECTIONS_NO_DIRECT_ACCESS | set([CONDITIONS])) RES_CONDITION = 'Condition' _RESOURCE_KEYS = CfnTemplateBase._RESOURCE_KEYS + (RES_CONDITION,) HOT_TO_CFN_RES_ATTRS = CfnTemplateBase.HOT_TO_CFN_RES_ATTRS HOT_TO_CFN_RES_ATTRS.update({'condition': RES_CONDITION}) OUTPUT_CONDITION = 'Condition' OUTPUT_KEYS = CfnTemplateBase.OUTPUT_KEYS + (OUTPUT_CONDITION,) functions = { 'Fn::FindInMap': cfn_funcs.FindInMap, 'Fn::GetAZs': cfn_funcs.GetAZs, 'Ref': cfn_funcs.Ref, 'Fn::GetAtt': cfn_funcs.GetAtt, 'Fn::Select': cfn_funcs.Select, 'Fn::Join': cfn_funcs.Join, 'Fn::Split': cfn_funcs.Split, 'Fn::Replace': cfn_funcs.Replace, 'Fn::Base64': cfn_funcs.Base64, 'Fn::MemberListToMap': cfn_funcs.MemberListToMap, 'Fn::ResourceFacade': cfn_funcs.ResourceFacade, 'Fn::If': cfn_funcs.If, } condition_functions = { 'Fn::Equals': cfn_funcs.Equals, 'Ref': cfn_funcs.ParamRef, 'Fn::FindInMap': cfn_funcs.FindInMap, 'Fn::Not': cfn_funcs.Not, 'Fn::And': cfn_funcs.And, 'Fn::Or': cfn_funcs.Or } def __init__(self, tmpl, template_id=None, files=None, env=None): super(CfnTemplate, self).__init__(tmpl, template_id, files, env) self.merge_sections = [self.PARAMETERS, self.CONDITIONS] def _get_condition_definitions(self): return self.t.get(self.CONDITIONS, {}) def _rsrc_defn_args(self, stack, name, data): for arg in super(CfnTemplate, self)._rsrc_defn_args(stack, name, data): yield arg parse_cond = functools.partial(self.parse_condition, stack) yield ('condition', self._parse_resource_field(self.RES_CONDITION, (str, bool, function.Function), 'string or boolean', name, data, parse_cond))
[docs]class HeatTemplate(CfnTemplateBase): functions = { 'Fn::FindInMap': cfn_funcs.FindInMap, 'Fn::GetAZs': cfn_funcs.GetAZs, 'Ref': cfn_funcs.Ref, 'Fn::GetAtt': cfn_funcs.GetAtt, 'Fn::Select': cfn_funcs.Select, 'Fn::Join': cfn_funcs.Join, 'Fn::Split': cfn_funcs.Split, 'Fn::Replace': cfn_funcs.Replace, 'Fn::Base64': cfn_funcs.Base64, 'Fn::MemberListToMap': cfn_funcs.MemberListToMap, 'Fn::ResourceFacade': cfn_funcs.ResourceFacade, }