_resolve_dependencies() - Code Metrics - Inspection of "Update changelog." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( c278f1...4c6d50 )

by Plexxi

created 2016-06-26 06:42 UTC

_resolve_dependencies() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
c	0
b	0
f	0
dl	0
loc	22
rs	7.7857

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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 six
import networkx as nx

from jinja2 import meta
from st2common import log as logging
from st2common.constants.action import ACTION_CONTEXT_KV_PREFIX
from st2common.constants.keyvalue import SYSTEM_SCOPE
from st2common.exceptions.param import ParamException
from st2common.services.keyvalues import KeyValueLookup
from st2common.util.casts import get_cast
from st2common.util.compat import to_unicode
from st2common.util import jinja as jinja_utils


LOG = logging.getLogger(__name__)
ENV = jinja_utils.get_jinja_environment()

__all__ = [
    'render_live_params',
    'render_final_params',
]


def _split_params(runner_parameters, action_parameters, mixed_params):
    def pf(params, skips):
        result = {k: v for k, v in six.iteritems(mixed_params)
                  if k in params and k not in skips}
        return result
    return (pf(runner_parameters, {}), pf(action_parameters, runner_parameters))


def _cast_params(rendered, parameter_schemas):
    '''
    It's just here to make tests happy
    '''
    casted_params = {}
    for k, v in six.iteritems(rendered):
        casted_params[k] = _cast(v, parameter_schemas[k] or {})
    return casted_params


def _cast(v, parameter_schema):
    if v is None or not parameter_schema:
        return v

    parameter_type = parameter_schema.get('type', None)
    if not parameter_type:
        return v

    cast = get_cast(cast_type=parameter_type)
    if not cast:
        return v

    return cast(v)


def _create_graph(action_context):
    '''
    Creates a generic directed graph for depencency tree and fills it with basic context variables
    '''
    G = nx.DiGraph()
    G.add_node(SYSTEM_SCOPE, value=KeyValueLookup(scope=SYSTEM_SCOPE))
    G.add_node(ACTION_CONTEXT_KV_PREFIX, value=action_context)
    return G


def _process(G, name, value):
    '''
    Determines whether parameter is a template or a value. Adds graph nodes and edges accordingly.
    '''
    # Jinja defaults to ascii parser in python 2.x unless you set utf-8 support on per module level
    # Instead we're just assuming every string to be a unicode string
    if isinstance(value, str):
        value = to_unicode(value)
    template_ast = ENV.parse(value)
    # Dependencies of the node represent jinja variables used in the template
    # We're connecting nodes with an edge for every depencency to traverse them in the right order
    # and also make sure that we don't have missing or cyclic dependencies upfront.
    dependencies = meta.find_undeclared_variables(template_ast)
    if dependencies:
        G.add_node(name, template=value)
        for dependency in dependencies:
            G.add_edge(dependency, name)
    else:
        G.add_node(name, value=value)


def _process_defaults(G, schemas):
    '''
    Process dependencies for parameters default values in the order schemas are defined.
    '''
    for schema in schemas:
        for name, value in six.iteritems(schema):
            absent = name not in G.node
            is_none = G.node.get(name, {}).get('value') is None
            immutable = value.get('immutable', False)
            if absent or is_none or immutable:
                _process(G, name, value.get('default'))


def _validate(G):
    '''
    Validates dependency graph to ensure it has no missing or cyclic dependencies
    '''
    for name in G.nodes():
        if 'value' not in G.node[name] and 'template' not in G.node[name]:
            msg = 'Dependecy unsatisfied in %s' % name
            raise ParamException(msg)

    if not nx.is_directed_acyclic_graph(G):
        msg = 'Cyclic dependecy found'
        raise ParamException(msg)


def _render(node, render_context):
    '''
    Render the node depending on its type
    '''
    if 'template' in node:
        return ENV.from_string(node['template']).render(render_context)
    if 'value' in node:
        return node['value']


def _resolve_dependencies(G):
    '''
    Traverse the dependency graph starting from resolved nodes
    '''
    context = {}
    for name in nx.topological_sort(G):
        node = G.node[name]
        try:
            if 'template' in node and isinstance(node.get('template', None), list):
                rendered_list = list()
                for template in G.node[name]['template']:
                    rendered_list.append(
                        _render(dict(template=template), context)
                    )
                context[name] = rendered_list
            else:
                context[name] = _render(node, context)
        except Exception as e:
            LOG.debug('Failed to render %s: %s', name, e, exc_info=True)
            msg = 'Failed to render parameter "%s": %s' % (name, str(e))
            raise ParamException(msg)
    return context


def _cast_params_from(params, context, schemas):
    '''
    Pick a list of parameters from context and cast each of them according to the schemas provided
    '''
    result = {}
    for name in params:
        param_schema = {}
        for schema in schemas:
            if name in schema:
                param_schema = schema[name]
        result[name] = _cast(context[name], param_schema)
    return result


def render_live_params(runner_parameters, action_parameters, params, action_context):
    '''
    Renders list of parameters. Ensures that there's no cyclic or missing dependencies. Returns a
    dict of plain rendered parameters.
    '''
    G = _create_graph(action_context)

    [_process(G, name, value) for name, value in six.iteritems(params)]
    _process_defaults(G, [action_parameters, runner_parameters])
    _validate(G)

    context = _resolve_dependencies(G)
    live_params = _cast_params_from(params, context, [action_parameters, runner_parameters])

    return live_params


def render_final_params(runner_parameters, action_parameters, params, action_context):
    '''
    Renders missing parameters required for action to execute. Treats parameters from the dict as
    plain values instead of trying to render them again. Returns dicts for action and runner
    parameters.
    '''
    G = _create_graph(action_context)

    # by that point, all params should already be resolved so any template should be treated value
    [G.add_node(name, value=value) for name, value in six.iteritems(params)]
    _process_defaults(G, [action_parameters, runner_parameters])
    _validate(G)

    context = _resolve_dependencies(G)
    context = _cast_params_from(context, context, [action_parameters, runner_parameters])

    return _split_params(runner_parameters, action_parameters, context)


def get_finalized_params(runnertype_parameter_info, action_parameter_info, liveaction_parameters,
                         action_context):
    '''
    Left here to keep tests running. Later we would need to split tests so they start testing each
    function separately.
    '''
    params = render_live_params(runnertype_parameter_info, action_parameter_info,
                                liveaction_parameters, action_context)
    return render_final_params(runnertype_parameter_info, action_parameter_info, params,
                               action_context)


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import six
17			import networkx as nx
18
19			from jinja2 import meta
20			from st2common import log as logging
21			from st2common.constants.action import ACTION_CONTEXT_KV_PREFIX
22			from st2common.constants.keyvalue import SYSTEM_SCOPE
23			from st2common.exceptions.param import ParamException
24			from st2common.services.keyvalues import KeyValueLookup
25			from st2common.util.casts import get_cast
26			from st2common.util.compat import to_unicode
27			from st2common.util import jinja as jinja_utils
28
29
30			LOG = logging.getLogger(__name__)
31			ENV = jinja_utils.get_jinja_environment()
32
33			__all__ = [
34			'render_live_params',
35			'render_final_params',
36			]
37
38
39			def _split_params(runner_parameters, action_parameters, mixed_params):
40			def pf(params, skips):
41			result = {k: v for k, v in six.iteritems(mixed_params)
42			if k in params and k not in skips}
43			return result
44			return (pf(runner_parameters, {}), pf(action_parameters, runner_parameters))
45
46
47			def _cast_params(rendered, parameter_schemas):
48			'''
49			It's just here to make tests happy
50			'''
51			casted_params = {}
52			for k, v in six.iteritems(rendered):
53			casted_params[k] = _cast(v, parameter_schemas[k] or {})
54			return casted_params
55
56
57			def _cast(v, parameter_schema):
58			if v is None or not parameter_schema:
59			return v
60
61			parameter_type = parameter_schema.get('type', None)
62			if not parameter_type:
63			return v
64
65			cast = get_cast(cast_type=parameter_type)
66			if not cast:
67			return v
68
69			return cast(v)
70
71
72			def _create_graph(action_context):
73			'''
74			Creates a generic directed graph for depencency tree and fills it with basic context variables
75			'''
76			G = nx.DiGraph()
77			G.add_node(SYSTEM_SCOPE, value=KeyValueLookup(scope=SYSTEM_SCOPE))
78			G.add_node(ACTION_CONTEXT_KV_PREFIX, value=action_context)
79			return G
80
81
82			def _process(G, name, value):
83			'''
84			Determines whether parameter is a template or a value. Adds graph nodes and edges accordingly.
85			'''
86			# Jinja defaults to ascii parser in python 2.x unless you set utf-8 support on per module level
87			# Instead we're just assuming every string to be a unicode string
88			if isinstance(value, str):
89			value = to_unicode(value)
90			template_ast = ENV.parse(value)
91			# Dependencies of the node represent jinja variables used in the template
92			# We're connecting nodes with an edge for every depencency to traverse them in the right order
93			# and also make sure that we don't have missing or cyclic dependencies upfront.
94			dependencies = meta.find_undeclared_variables(template_ast)
95			if dependencies:
96			G.add_node(name, template=value)
97			for dependency in dependencies:
98			G.add_edge(dependency, name)
99			else:
100			G.add_node(name, value=value)
101
102
103			def _process_defaults(G, schemas):
104			'''
105			Process dependencies for parameters default values in the order schemas are defined.
106			'''
107			for schema in schemas:
108			for name, value in six.iteritems(schema):
109			absent = name not in G.node
110			is_none = G.node.get(name, {}).get('value') is None
111			immutable = value.get('immutable', False)
112			if absent or is_none or immutable:
113			_process(G, name, value.get('default'))
114
115
116			def _validate(G):
117			'''
118			Validates dependency graph to ensure it has no missing or cyclic dependencies
119			'''
120			for name in G.nodes():
121			if 'value' not in G.node[name] and 'template' not in G.node[name]:
122			msg = 'Dependecy unsatisfied in %s' % name
123			raise ParamException(msg)
124
125			if not nx.is_directed_acyclic_graph(G):
126			msg = 'Cyclic dependecy found'
127			raise ParamException(msg)
128
129
130			def _render(node, render_context):
131			'''
132			Render the node depending on its type
133			'''
134			if 'template' in node:
135			return ENV.from_string(node['template']).render(render_context)
136			if 'value' in node:
137			return node['value']
138
139
140			def _resolve_dependencies(G):
141			'''
142			Traverse the dependency graph starting from resolved nodes
143			'''
144			context = {}
145			for name in nx.topological_sort(G):
146			node = G.node[name]
147			try:
148			if 'template' in node and isinstance(node.get('template', None), list):
149			rendered_list = list()
150			for template in G.node[name]['template']:
151			rendered_list.append(
152			_render(dict(template=template), context)
153			)
154			context[name] = rendered_list
155			else:
156			context[name] = _render(node, context)
157			except Exception as e:
158			LOG.debug('Failed to render %s: %s', name, e, exc_info=True)
159			msg = 'Failed to render parameter "%s": %s' % (name, str(e))
160			raise ParamException(msg)
161			return context
162
163
164			def _cast_params_from(params, context, schemas):
165			'''
166			Pick a list of parameters from context and cast each of them according to the schemas provided
167			'''
168			result = {}
169			for name in params:
170			param_schema = {}
171			for schema in schemas:
172			if name in schema:
173			param_schema = schema[name]
174			result[name] = _cast(context[name], param_schema)
175			return result
176
177
178			def render_live_params(runner_parameters, action_parameters, params, action_context):
179			'''
180			Renders list of parameters. Ensures that there's no cyclic or missing dependencies. Returns a
181			dict of plain rendered parameters.
182			'''
183			G = _create_graph(action_context)
184
185			[_process(G, name, value) for name, value in six.iteritems(params)]
186			_process_defaults(G, [action_parameters, runner_parameters])
187			_validate(G)
188
189			context = _resolve_dependencies(G)
190			live_params = _cast_params_from(params, context, [action_parameters, runner_parameters])
191
192			return live_params
193
194
195			def render_final_params(runner_parameters, action_parameters, params, action_context):
196			'''
197			Renders missing parameters required for action to execute. Treats parameters from the dict as
198			plain values instead of trying to render them again. Returns dicts for action and runner
199			parameters.
200			'''
201			G = _create_graph(action_context)
202
203			# by that point, all params should already be resolved so any template should be treated value
204			[G.add_node(name, value=value) for name, value in six.iteritems(params)]
205			_process_defaults(G, [action_parameters, runner_parameters])
206			_validate(G)
207
208			context = _resolve_dependencies(G)
209			context = _cast_params_from(context, context, [action_parameters, runner_parameters])
210
211			return _split_params(runner_parameters, action_parameters, context)
212
213
214			def get_finalized_params(runnertype_parameter_info, action_parameter_info, liveaction_parameters,
215			action_context):
216			'''
217			Left here to keep tests running. Later we would need to split tests so they start testing each
218			function separately.
219			'''
220			params = render_live_params(runnertype_parameter_info, action_parameter_info,
221			liveaction_parameters, action_context)
222			return render_final_params(runnertype_parameter_info, action_parameter_info, params,
223			action_context)
224

Plexxi / st2

GitHub Access Token became invalid

Push — develop ( c278f1...4c6d50 )

_resolve_dependencies() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like