convert_jinja_to_raw_block() - Code Metrics - Inspection of "Fix invocation of Mistral workflow from Action Cha..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#3609)

by W

created 2017-07-26 17:28 UTC

convert_jinja_to_raw_block() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	8
c	1
b	0
f	0
dl	0
loc	12
rs	6.6666

# 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 json
import re
import six

from st2common import log as logging
from st2common.util.compat import to_unicode


__all__ = [
    'get_jinja_environment',
    'render_values',
    'is_jinja_expression'
]

# Magic string to which None type is serialized when using use_none filter
NONE_MAGIC_VALUE = '%*****__%NONE%__*****%'

JINJA_EXPRESSIONS_START_MARKERS = [
    '{{',
    '{%'
]

JINJA_REGEX = '({{(.*)}})'
JINJA_REGEX_PTRN = re.compile(JINJA_REGEX)
JINJA_BLOCK_REGEX = '({%(.*)%})'
JINJA_BLOCK_REGEX_PTRN = re.compile(JINJA_BLOCK_REGEX)


LOG = logging.getLogger(__name__)


def use_none(value):
    if value is None:
        return NONE_MAGIC_VALUE

    return value


def get_filters():
    # Lazy / late import to avoid long module import times
    from st2common.jinja.filters import crypto
    from st2common.jinja.filters import data
    from st2common.jinja.filters import regex
    from st2common.jinja.filters import complex_type
    from st2common.jinja.filters import time
    from st2common.jinja.filters import version
    from st2common.jinja.filters import json_escape

    return {
        'decrypt_kv': crypto.decrypt_kv,
        'to_json_string': data.to_json_string,
        'to_yaml_string': data.to_yaml_string,

        'to_complex': complex_type.to_complex,

        'regex_match': regex.regex_match,
        'regex_replace': regex.regex_replace,
        'regex_search': regex.regex_search,
        'regex_substring': regex.regex_substring,

        'to_human_time_from_seconds': time.to_human_time_from_seconds,

        'version_compare': version.version_compare,
        'version_more_than': version.version_more_than,
        'version_less_than': version.version_less_than,
        'version_equal': version.version_equal,
        'version_match': version.version_match,
        'version_bump_major': version.version_bump_major,
        'version_bump_minor': version.version_bump_minor,
        'version_bump_patch': version.version_bump_patch,
        'version_strip_patch': version.version_strip_patch,
        'use_none': use_none,

        'json_escape': json_escape.json_escape
    }


def get_jinja_environment(allow_undefined=False, trim_blocks=True, lstrip_blocks=True):
    '''
    jinja2.Environment object that is setup with right behaviors and custom filters.

    :param strict_undefined: If should allow undefined variables in templates
    :type strict_undefined: ``bool``

    '''
    # Late import to avoid very expensive in-direct import (~1 second) when this function
    # is not called / used
    import jinja2

    undefined = jinja2.Undefined if allow_undefined else jinja2.StrictUndefined
    env = jinja2.Environment(  # nosec
        undefined=undefined,
        trim_blocks=trim_blocks,
        lstrip_blocks=lstrip_blocks
    )
    env.filters.update(get_filters())
    env.tests['in'] = lambda item, list: item in list
    return env


def render_values(mapping=None, context=None, allow_undefined=False):
    """
    Render an incoming mapping using context provided in context using Jinja2. Returns a dict
    containing rendered mapping.

    :param mapping: Input as a dictionary of key value pairs.
    :type mapping: ``dict``

    :param context: Context to be used for dictionary.
    :type context: ``dict``

    :rtype: ``dict``
    """

    if not context or not mapping:
        return mapping

    # Add in special __context variable that provides an easy way to get access to entire context.
    # This mean __context is a reserve key word although backwards compat is preserved by making
    # sure that real context is updated later and therefore will override the __context value.
    super_context = {}
    super_context['__context'] = context
    super_context.update(context)

    env = get_jinja_environment(allow_undefined=allow_undefined)
    rendered_mapping = {}
    for k, v in six.iteritems(mapping):
        # jinja2 works with string so transform list and dict to strings.
        reverse_json_dumps = False
        if isinstance(v, dict) or isinstance(v, list):
            v = json.dumps(v)
            reverse_json_dumps = True
        else:
            # Special case for text type to handle unicode
            if isinstance(v, six.string_types):
                v = to_unicode(v)
            else:
                # Other types (e.g. boolean, etc.)
                v = str(v)

        try:
            LOG.info('Rendering string %s. Super context=%s', v, super_context)
            rendered_v = env.from_string(v).render(super_context)
        except Exception as e:
            # Attach key and value which failed the rendering
            e.key = k
            e.value = v
            raise e

        # no change therefore no templatization so pick params from original to retain
        # original type
        if rendered_v == v:
            rendered_mapping[k] = mapping[k]
            continue
        if reverse_json_dumps:
            rendered_v = json.loads(rendered_v)
        rendered_mapping[k] = rendered_v
    LOG.info('Mapping: %s, rendered_mapping: %s, context: %s', mapping, rendered_mapping, context)
    return rendered_mapping


def is_jinja_expression(value):
    """
    Function which very simplisticly detect if the provided value contains or is a Jinja
    expression.
    """
    if not value or not isinstance(value, six.string_types):
        return False

    for marker in JINJA_EXPRESSIONS_START_MARKERS:
        if marker in value:
            return True

    return False


def convert_jinja_to_raw_block(value):
    if isinstance(value, dict):
        return {k: convert_jinja_to_raw_block(v) for k, v in six.iteritems(value)}

    if isinstance(value, list):
        return [convert_jinja_to_raw_block(v) for v in value]

    if isinstance(value, six.string_types):
        if JINJA_REGEX_PTRN.findall(value) or JINJA_BLOCK_REGEX_PTRN.findall(value):
            return '{% raw %}' + value + '{% endraw %}'

    return value


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 json
17			import re
18			import six
19
20			from st2common import log as logging
21			from st2common.util.compat import to_unicode
22
23
24			__all__ = [
25			'get_jinja_environment',
26			'render_values',
27			'is_jinja_expression'
28			]
29
30			# Magic string to which None type is serialized when using use_none filter
31			NONE_MAGIC_VALUE = '%***__%NONE%__***%'
32
33			JINJA_EXPRESSIONS_START_MARKERS = [
34			'{{',
35			'{%'
36			]
37
38			JINJA_REGEX = '({{(.*)}})'
39			JINJA_REGEX_PTRN = re.compile(JINJA_REGEX)
40			JINJA_BLOCK_REGEX = '({%(.*)%})'
41			JINJA_BLOCK_REGEX_PTRN = re.compile(JINJA_BLOCK_REGEX)
42
43
44			LOG = logging.getLogger(__name__)
45
46
47			def use_none(value):
48			if value is None:
49			return NONE_MAGIC_VALUE
50
51			return value
52
53
54			def get_filters():
55			# Lazy / late import to avoid long module import times
56			from st2common.jinja.filters import crypto
57			from st2common.jinja.filters import data
58			from st2common.jinja.filters import regex
59			from st2common.jinja.filters import complex_type
60			from st2common.jinja.filters import time
61			from st2common.jinja.filters import version
62			from st2common.jinja.filters import json_escape
63
64			return {
65			'decrypt_kv': crypto.decrypt_kv,
66			'to_json_string': data.to_json_string,
67			'to_yaml_string': data.to_yaml_string,
68
69			'to_complex': complex_type.to_complex,
70
71			'regex_match': regex.regex_match,
72			'regex_replace': regex.regex_replace,
73			'regex_search': regex.regex_search,
74			'regex_substring': regex.regex_substring,
75
76			'to_human_time_from_seconds': time.to_human_time_from_seconds,
77
78			'version_compare': version.version_compare,
79			'version_more_than': version.version_more_than,
80			'version_less_than': version.version_less_than,
81			'version_equal': version.version_equal,
82			'version_match': version.version_match,
83			'version_bump_major': version.version_bump_major,
84			'version_bump_minor': version.version_bump_minor,
85			'version_bump_patch': version.version_bump_patch,
86			'version_strip_patch': version.version_strip_patch,
87			'use_none': use_none,
88
89			'json_escape': json_escape.json_escape
90			}
91
92
93			def get_jinja_environment(allow_undefined=False, trim_blocks=True, lstrip_blocks=True):
94			'''
95			jinja2.Environment object that is setup with right behaviors and custom filters.
96
97			:param strict_undefined: If should allow undefined variables in templates
98			:type strict_undefined: ``bool``
99
100			'''
101			# Late import to avoid very expensive in-direct import (~1 second) when this function
102			# is not called / used
103			import jinja2
104
105			undefined = jinja2.Undefined if allow_undefined else jinja2.StrictUndefined
106			env = jinja2.Environment( # nosec
107			undefined=undefined,
108			trim_blocks=trim_blocks,
109			lstrip_blocks=lstrip_blocks
110			)
111			env.filters.update(get_filters())
112			env.tests['in'] = lambda item, list: item in list
113			return env
114
115
116			def render_values(mapping=None, context=None, allow_undefined=False):
117			"""
118			Render an incoming mapping using context provided in context using Jinja2. Returns a dict
119			containing rendered mapping.
120
121			:param mapping: Input as a dictionary of key value pairs.
122			:type mapping: ``dict``
123
124			:param context: Context to be used for dictionary.
125			:type context: ``dict``
126
127			:rtype: ``dict``
128			"""
129
130			if not context or not mapping:
131			return mapping
132
133			# Add in special __context variable that provides an easy way to get access to entire context.
134			# This mean __context is a reserve key word although backwards compat is preserved by making
135			# sure that real context is updated later and therefore will override the __context value.
136			super_context = {}
137			super_context['__context'] = context
138			super_context.update(context)
139
140			env = get_jinja_environment(allow_undefined=allow_undefined)
141			rendered_mapping = {}
142			for k, v in six.iteritems(mapping):
143			# jinja2 works with string so transform list and dict to strings.
144			reverse_json_dumps = False
145			if isinstance(v, dict) or isinstance(v, list):
146			v = json.dumps(v)
147			reverse_json_dumps = True
148			else:
149			# Special case for text type to handle unicode
150			if isinstance(v, six.string_types):
151			v = to_unicode(v)
152			else:
153			# Other types (e.g. boolean, etc.)
154			v = str(v)
155
156			try:
157			LOG.info('Rendering string %s. Super context=%s', v, super_context)
158			rendered_v = env.from_string(v).render(super_context)
159			except Exception as e:
160			# Attach key and value which failed the rendering
161			e.key = k
162			e.value = v
163			raise e
164
165			# no change therefore no templatization so pick params from original to retain
166			# original type
167			if rendered_v == v:
168			rendered_mapping[k] = mapping[k]
169			continue
170			if reverse_json_dumps:
171			rendered_v = json.loads(rendered_v)
172			rendered_mapping[k] = rendered_v
173			LOG.info('Mapping: %s, rendered_mapping: %s, context: %s', mapping, rendered_mapping, context)
174			return rendered_mapping
175
176
177			def is_jinja_expression(value):
178			"""
179			Function which very simplisticly detect if the provided value contains or is a Jinja
180			expression.
181			"""
182			if not value or not isinstance(value, six.string_types):
183			return False
184
185			for marker in JINJA_EXPRESSIONS_START_MARKERS:
186			if marker in value:
187			return True
188
189			return False
190
191
192			def convert_jinja_to_raw_block(value):
193			if isinstance(value, dict):
194			return {k: convert_jinja_to_raw_block(v) for k, v in six.iteritems(value)}
195
196			if isinstance(value, list):
197			return [convert_jinja_to_raw_block(v) for v in value]
198
199			if isinstance(value, six.string_types):
200			if JINJA_REGEX_PTRN.findall(value) or JINJA_BLOCK_REGEX_PTRN.findall(value):
201			return '{% raw %}' + value + '{% endraw %}'
202
203			return value
204

StackStorm / st2

Pull Request — master (#3609)

convert_jinja_to_raw_block() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like