Source code for heat.engine.attributes

#
#    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 collections
import functools

from oslo_utils import strutils

from heat.common.i18n import _
from heat.engine import constraints as constr
from heat.engine import support

from oslo_log import log as logging

LOG = logging.getLogger(__name__)


[docs]class Schema(constr.Schema):
    """Simple schema class for attributes.

    Schema objects are serializable to dictionaries following a superset of
    the HOT input Parameter schema using dict().
    """

    KEYS = (
        DESCRIPTION, TYPE
    ) = (
        'description', 'type',
    )

    CACHE_MODES = (
        CACHE_LOCAL,
        CACHE_NONE
    ) = (
        'cache_local',
        'cache_none'
    )

    TYPES = (
        UNKNOWN, STRING, MAP, LIST, INTEGER, BOOLEAN
    ) = (
        None, 'String', 'Map', 'List', 'Integer', 'Boolean'
    )

    def __init__(self, description=None,
                 support_status=support.SupportStatus(),
                 cache_mode=CACHE_LOCAL,
                 type=UNKNOWN):
        super(Schema, self).__init__(type, description)
        self.support_status = support_status
        self.cache_mode = cache_mode
        self.validate()

    def __getitem__(self, key):
        if key == self.DESCRIPTION:
            if self.description is not None:
                return self.description

        elif key == self.TYPE:
            if self.type is not None:
                return self.type.lower()

        raise KeyError(key)

[docs]    @classmethod
    def from_attribute(cls, schema_dict):
        """Return a Property Schema corresponding to a Attribute Schema."""
        msg = 'Old attribute schema is not supported'
        assert isinstance(schema_dict, cls), msg
        return schema_dict


[docs]def schemata(schema):
    """Return dictionary of Schema objects for given dictionary of schemata."""
    return dict((n, Schema.from_attribute(s)) for n, s in schema.items())


[docs]class Attribute(object):
    """An Attribute schema."""

    def __init__(self, attr_name, schema):
        """Initialise with a name and schema.

        :param attr_name: the name of the attribute
        :param schema: attribute schema
        """
        self.name = attr_name
        self.schema = Schema.from_attribute(schema)

[docs]    def support_status(self):
        return self.schema.support_status

[docs]    def as_output(self, resource_name, template_type='cfn'):
        """Output entry for a provider template with the given resource name.

        :param resource_name: the logical name of the provider resource
        :param template_type: the template type to generate
        :returns: This attribute as a template 'Output' entry for
                  cfn template and 'output' entry for hot template
        """
        if template_type == 'hot':
            return {
                "value": {"get_attr": [resource_name, self.name]},
                "description": self.schema.description
            }
        else:
            return {
                "Value": {"Fn::GetAtt": [resource_name, self.name]},
                "Description": self.schema.description
            }


def _stack_id_output(resource_name, template_type='cfn'):
    if template_type == 'hot':
        return {
            "value": {"get_resource": resource_name},
        }
    else:
        return {
            "Value": {"Ref": resource_name},
        }


BASE_ATTRIBUTES = (SHOW_ATTR, ) = ('show', )

# Returned by function.dep_attrs() to indicate that all attributes are
# referenced
ALL_ATTRIBUTES = '*'


[docs]class Attributes(collections.abc.Mapping):
    """Models a collection of Resource Attributes."""

    def __init__(self, res_name, schema, resolver):
        self._resource_name = res_name
        self._resolver = resolver
        self.set_schema(schema)
        self.reset_resolved_values()

        assert ALL_ATTRIBUTES not in schema, \
            "Invalid attribute name '%s'" % ALL_ATTRIBUTES

[docs]    def reset_resolved_values(self):
        if hasattr(self, '_resolved_values'):
            self._has_new_resolved = len(self._resolved_values) > 0
        else:
            self._has_new_resolved = False
        self._resolved_values = {}

[docs]    def set_schema(self, schema):
        self._attributes = self._make_attributes(schema)

[docs]    def get_cache_mode(self, attribute_name):
        """Return the cache mode for the specified attribute.

        If the attribute is not defined in the schema, the default cache
        mode (CACHE_LOCAL) is returned.
        """
        try:
            return self._attributes[attribute_name].schema.cache_mode
        except KeyError:
            return Schema.CACHE_LOCAL

    @staticmethod
    def _make_attributes(schema):
        return dict((n, Attribute(n, d)) for n, d in schema.items())

[docs]    @staticmethod
    def as_outputs(resource_name, resource_class, template_type='cfn'):
        """Dict of Output entries for a provider template with resource name.

        :param resource_name: logical name of the resource
        :param resource_class: resource implementation class
        :returns: The attributes of the specified resource_class as a template
                  Output map
        """
        attr_schema = {}
        for name, schema_data in resource_class.attributes_schema.items():
            schema = Schema.from_attribute(schema_data)
            if schema.support_status.status != support.HIDDEN:
                attr_schema[name] = schema
        attr_schema.update(resource_class.base_attributes_schema)
        attribs = Attributes._make_attributes(attr_schema).items()

        outp = dict((n, att.as_output(resource_name,
                                      template_type)) for n, att in attribs)
        outp['OS::stack_id'] = _stack_id_output(resource_name, template_type)
        return outp

[docs]    @staticmethod
    def schema_from_outputs(json_snippet):
        if json_snippet:
            return dict((k, Schema(v.get("Description")))
                        for k, v in json_snippet.items())
        return {}

    def _validate_type(self, attrib, value):
        if attrib.schema.type == attrib.schema.STRING:
            if not isinstance(value, str):
                LOG.warning("Attribute %(name)s is not of type "
                            "%(att_type)s",
                            {'name': attrib.name,
                             'att_type': attrib.schema.STRING})
        elif attrib.schema.type == attrib.schema.LIST:
            if (not isinstance(value, collections.abc.Sequence)
                    or isinstance(value, str)):
                LOG.warning("Attribute %(name)s is not of type "
                            "%(att_type)s",
                            {'name': attrib.name,
                             'att_type': attrib.schema.LIST})
        elif attrib.schema.type == attrib.schema.MAP:
            if not isinstance(value, collections.abc.Mapping):
                LOG.warning("Attribute %(name)s is not of type "
                            "%(att_type)s",
                            {'name': attrib.name,
                             'att_type': attrib.schema.MAP})
        elif attrib.schema.type == attrib.schema.INTEGER:
            if not isinstance(value, int):
                LOG.warning("Attribute %(name)s is not of type "
                            "%(att_type)s",
                            {'name': attrib.name,
                             'att_type': attrib.schema.INTEGER})
        elif attrib.schema.type == attrib.schema.BOOLEAN:
            try:
                strutils.bool_from_string(value, strict=True)
            except ValueError:
                LOG.warning("Attribute %(name)s is not of type "
                            "%(att_type)s",
                            {'name': attrib.name,
                             'att_type': attrib.schema.BOOLEAN})

    @property
    def cached_attrs(self):
        return self._resolved_values

    @cached_attrs.setter
    def cached_attrs(self, c_attrs):
        if c_attrs is None:
            self._resolved_values = {}
        else:
            self._resolved_values = c_attrs
        self._has_new_resolved = False

[docs]    def set_cached_attr(self, key, value):
        self._resolved_values[key] = value
        self._has_new_resolved = True

[docs]    def has_new_cached_attrs(self):
        """Returns True if cached_attrs have changed

        Allows the caller to determine if this instance's cached_attrs
        have been updated since they were initially set (if at all).
        """
        return self._has_new_resolved

    def __getitem__(self, key):
        if key not in self:
            raise KeyError(_('%(resource)s: Invalid attribute %(key)s') %
                           dict(resource=self._resource_name, key=key))

        attrib = self._attributes.get(key)
        if attrib.schema.cache_mode == Schema.CACHE_NONE:
            return self._resolver(key)

        if key in self._resolved_values:
            return self._resolved_values[key]

        value = self._resolver(key)

        if value is not None:
            # validate the value against its type
            self._validate_type(attrib, value)
            # only store if not None, it may resolve to an actual value
            # on subsequent calls
            self.set_cached_attr(key, value)
        return value

    def __len__(self):
        return len(self._attributes)

    def __contains__(self, key):
        return key in self._attributes

    def __iter__(self):
        return iter(self._attributes)

    def __repr__(self):
        return ("Attributes for %s:\n\t" % self._resource_name +
                '\n\t'.join(self.values()))


[docs]def select_from_attribute(attribute_value, path):
    """Select an element from an attribute value.

    :param attribute_value: the attribute value.
    :param path: a list of path components to select from the attribute.
    :returns: the selected attribute component value.
    """
    def get_path_component(collection, key):
        if not isinstance(collection, (collections.abc.Mapping,
                                       collections.abc.Sequence)):
            raise TypeError(_("Can't traverse attribute path"))

        if not isinstance(key, (str, int)):
            raise TypeError(_('Path components in attributes must be strings'))

        return collection[key]

    try:
        return functools.reduce(get_path_component, path, attribute_value)
    except (KeyError, IndexError, TypeError):
        return None