#
# 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 abc
import collections
import itertools
import weakref
from heat.common import exception
from heat.common.i18n import _
[docs]
class Function(metaclass=abc.ABCMeta):
"""Abstract base class for template functions."""
def __init__(self, stack, fn_name, args):
"""Initialise with a Stack, the function name and the arguments.
All functions take the form of a single-item map in JSON::
{ <fn_name> : <args> }
"""
super(Function, self).__init__()
self._stackref = weakref.ref(stack) if stack is not None else None
self.fn_name = fn_name
self.args = args
@property
def stack(self):
ref = self._stackref
if ref is None:
return None
stack = ref()
assert stack is not None, ("Need a reference to the "
"StackDefinition object")
return stack
[docs]
def validate(self):
"""Validate arguments without resolving the function.
Function subclasses must override this method to validate their
args.
"""
validate(self.args)
[docs]
@abc.abstractmethod
def result(self):
"""Return the result of resolving the function.
Function subclasses must override this method to calculate their
results.
"""
return {self.fn_name: self.args}
[docs]
def dependencies(self, path):
return dependencies(self.args, '.'.join([path, self.fn_name]))
[docs]
def dep_attrs(self, resource_name):
"""Return the attributes of the specified resource that are referenced.
Return an iterator over any attributes of the specified resource that
this function references.
The special value heat.engine.attributes.ALL_ATTRIBUTES may be used to
indicate that all attributes of the resource are required.
"""
return dep_attrs(self.args, resource_name)
[docs]
def all_dep_attrs(self):
"""Return resource, attribute name pairs of all attributes referenced.
Return an iterator over the resource name, attribute name tuples of
all attributes that this function references.
The special value heat.engine.attributes.ALL_ATTRIBUTES may be used to
indicate that all attributes of the resource are required.
By default this calls the dep_attrs() method, but subclasses can
override to provide a more efficient implementation.
"""
# If we are using the default dep_attrs method then it will only
# return data from the args anyway
if type(self).dep_attrs == Function.dep_attrs:
return all_dep_attrs(self.args)
def res_dep_attrs(resource_name):
return zip(itertools.repeat(resource_name),
self.dep_attrs(resource_name))
resource_names = self.stack.enabled_rsrc_names()
return itertools.chain.from_iterable(map(res_dep_attrs,
resource_names))
def __reduce__(self):
"""Return a representation of the function suitable for pickling.
This allows the copy module (which works by pickling and then
unpickling objects) to copy a template. Functions in the copy will
return to their original (JSON) form (i.e. a single-element map).
"""
return dict, ([(self.fn_name, self.args)],)
def _repr_result(self):
try:
return repr(self.result())
except (TypeError, ValueError):
return '???'
def __repr__(self):
"""Return a string representation of the function.
The representation includes the function name, arguments and result
(if available), as well as the name of the function class.
"""
fntype = type(self)
classname = '.'.join(filter(None,
(getattr(fntype,
attr,
'') for attr in ('__module__',
'__name__'))))
return '<%s {%s: %r} -> %s>' % (classname,
self.fn_name, self.args,
self._repr_result())
def __eq__(self, other):
"""Compare the result of this function for equality."""
try:
result = self.result()
if isinstance(other, Function):
return result == other.result()
else:
return result == other
except (TypeError, ValueError):
return NotImplemented
def __ne__(self, other):
"""Compare the result of this function for inequality."""
eq = self.__eq__(other)
if eq is NotImplemented:
return NotImplemented
return not eq
__hash__ = None
[docs]
class Macro(Function, metaclass=abc.ABCMeta):
"""Abstract base class for template macros.
A macro differs from a function in that it controls how the template is
parsed. As such, it operates on the syntax tree itself, not on the parsed
output.
"""
def __init__(self, stack, fn_name, raw_args, parse_func, template):
"""Initialise with the argument syntax tree and parser function."""
super(Macro, self).__init__(stack, fn_name, raw_args)
self._tmplref = weakref.ref(template) if template is not None else None
self.parsed = self.parse_args(parse_func)
@property
def template(self):
ref = self._tmplref
if ref is None:
return None
tmpl = ref()
assert tmpl is not None, "Need a reference to the Template object"
return tmpl
[docs]
@abc.abstractmethod
def parse_args(self, parse_func):
"""Parse the macro using the supplied parsing function.
Macro subclasses should override this method to control parsing of
the arguments.
"""
return parse_func(self.args)
[docs]
def validate(self):
"""Validate arguments without resolving the result."""
validate(self.parsed)
[docs]
def result(self):
"""Return the resolved result of the macro contents."""
return resolve(self.parsed, nullable=True)
[docs]
def dependencies(self, path):
return dependencies(self.parsed, '.'.join([path, self.fn_name]))
[docs]
def dep_attrs(self, resource_name):
"""Return the attributes of the specified resource that are referenced.
Return an iterator over any attributes of the specified resource that
this function references.
The special value heat.engine.attributes.ALL_ATTRIBUTES may be used to
indicate that all attributes of the resource are required.
"""
return dep_attrs(self.parsed, resource_name)
[docs]
def all_dep_attrs(self):
"""Return resource, attribute name pairs of all attributes referenced.
Return an iterator over the resource name, attribute name tuples of
all attributes that this function references.
The special value heat.engine.attributes.ALL_ATTRIBUTES may be used to
indicate that all attributes of the resource are required.
By default this calls the dep_attrs() method, but subclasses can
override to provide a more efficient implementation.
"""
# If we are using the default dep_attrs method then it will only
# return data from the transformed parsed args anyway
if type(self).dep_attrs == Macro.dep_attrs:
return all_dep_attrs(self.parsed)
return super(Macro, self).all_dep_attrs()
def __reduce__(self):
"""Return a representation of the macro result suitable for pickling.
This allows the copy module (which works by pickling and then
unpickling objects) to copy a template. Functions in the copy will
return to their original (JSON) form (i.e. a single-element map).
Unlike other functions, macros are *not* preserved during a copy. The
the processed (but unparsed) output is returned in their place.
"""
if isinstance(self.parsed, Function):
return self.parsed.__reduce__()
if self.parsed is None:
return type(None), tuple()
if self.parsed is Ellipsis:
return type(Ellipsis), tuple()
return type(self.parsed), (self.parsed,)
def _repr_result(self):
return repr(self.parsed)
def _non_null_item(i):
k, v = i
return v is not Ellipsis
def _non_null_value(v):
return v is not Ellipsis
[docs]
def resolve(snippet, nullable=False):
if isinstance(snippet, Function):
result = snippet.result()
if not (nullable or _non_null_value(result)):
result = None
return result
if isinstance(snippet, collections.abc.Mapping):
return dict(filter(_non_null_item,
((k, resolve(v, nullable=True))
for k, v in snippet.items())))
elif (not isinstance(snippet, str) and
isinstance(snippet, collections.abc.Iterable)):
return list(filter(_non_null_value,
(resolve(v, nullable=True) for v in snippet)))
return snippet
[docs]
def validate(snippet, path=None):
if path is None:
path = []
elif isinstance(path, str):
path = [path]
if isinstance(snippet, Function):
try:
snippet.validate()
except AssertionError:
raise
except Exception as e:
raise exception.StackValidationFailed(
path=path + [snippet.fn_name],
message=str(e))
elif isinstance(snippet, collections.abc.Mapping):
for k, v in snippet.items():
validate(v, path + [k])
elif (not isinstance(snippet, str) and
isinstance(snippet, collections.abc.Iterable)):
basepath = list(path)
parent = basepath.pop() if basepath else ''
for i, v in enumerate(snippet):
validate(v, basepath + ['%s[%d]' % (parent, i)])
[docs]
def dependencies(snippet, path=''):
"""Return an iterator over Resource dependencies in a template snippet.
The snippet should be already parsed to insert Function objects where
appropriate.
"""
if isinstance(snippet, Function):
return snippet.dependencies(path)
elif isinstance(snippet, collections.abc.Mapping):
def mkpath(key):
return '.'.join([path, str(key)])
deps = (dependencies(value,
mkpath(key)) for key, value in snippet.items())
return itertools.chain.from_iterable(deps)
elif (not isinstance(snippet, str) and
isinstance(snippet, collections.abc.Iterable)):
def mkpath(idx):
return ''.join([path, '[%d]' % idx])
deps = (dependencies(value,
mkpath(i)) for i, value in enumerate(snippet))
return itertools.chain.from_iterable(deps)
else:
return []
[docs]
def dep_attrs(snippet, resource_name):
"""Iterator over dependent attrs of a resource in a template snippet.
The snippet should be already parsed to insert Function objects where
appropriate.
:returns: an iterator over the attributes of the specified resource that
are referenced in the template snippet.
"""
if isinstance(snippet, Function):
return snippet.dep_attrs(resource_name)
elif isinstance(snippet, collections.abc.Mapping):
attrs = (dep_attrs(val, resource_name) for val in snippet.values())
return itertools.chain.from_iterable(attrs)
elif (not isinstance(snippet, str) and
isinstance(snippet, collections.abc.Iterable)):
attrs = (dep_attrs(value, resource_name) for value in snippet)
return itertools.chain.from_iterable(attrs)
return []
[docs]
def all_dep_attrs(snippet):
"""Iterator over resource, attribute name pairs referenced in a snippet.
The snippet should be already parsed to insert Function objects where
appropriate.
:returns: an iterator over the resource name, attribute name tuples of all
attributes that are referenced in the template snippet.
"""
if isinstance(snippet, Function):
return snippet.all_dep_attrs()
elif isinstance(snippet, collections.abc.Mapping):
res_attrs = (all_dep_attrs(value) for value in snippet.values())
return itertools.chain.from_iterable(res_attrs)
elif (not isinstance(snippet, str) and
isinstance(snippet, collections.abc.Iterable)):
res_attrs = (all_dep_attrs(value) for value in snippet)
return itertools.chain.from_iterable(res_attrs)
return []
[docs]
class Invalid(Function):
"""A function for checking condition functions and to force failures.
This function is used to force failures for functions that are not
supported in condition definition.
"""
def __init__(self, stack, fn_name, args):
raise ValueError(_('The function "%s" '
'is invalid in this context') % fn_name)
[docs]
def result(self):
return super(Invalid, self).result()
