st2common.util.serialize_positional_argument() - Code Metrics - Inspection of "Merge pull request #2592 from StackStorm/positiona..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 1df8f4...7d872c )

by Plexxi

created 2016-03-25 06:42 UTC

st2common.util.serialize_positional_argument() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	12
dl	0
loc	30
rs	2.7855

How to fix Complexity

# 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 simplejson as json
from collections import OrderedDict

from mongoengine import ValidationError
import six

from st2common import log as logging
from st2common.constants.action import LIVEACTION_STATUSES
from st2common.exceptions.db import StackStormDBObjectNotFoundError
from st2common.persistence.action import Action
from st2common.persistence.liveaction import LiveAction
from st2common.persistence.runner import RunnerType

LOG = logging.getLogger(__name__)


__all__ = [
    'get_action_parameters_specs',
    'get_runnertype_by_id',
    'get_runnertype_by_name',
    'get_action_by_id',
    'get_action_by_ref',
    'get_liveaction_by_id',
    'update_liveaction_status',
    'serialize_positional_argument',
    'get_args'
]


def get_action_parameters_specs(action_ref):
    """
    Retrieve parameters specifications schema for the provided action reference.

    Note: This function returns a union of action and action runner parameters.

    :param action_ref: Action reference.
    :type action_ref: ``str``

    :rtype: ``dict``
    """
    action_db = get_action_by_ref(ref=action_ref)

    parameters = {}
    if not action_db:
        return parameters

    runner_type_name = action_db.runner_type['name']
    runner_type_db = get_runnertype_by_name(runnertype_name=runner_type_name)

    # Runner type parameters should be added first before the action parameters.
    parameters.update(runner_type_db['runner_parameters'])
    parameters.update(action_db.parameters)

    return parameters


def get_runnertype_by_id(runnertype_id):
    """
        Get RunnerType by id.

        On error, raise StackStormDBObjectNotFoundError
    """
    try:
        runnertype = RunnerType.get_by_id(runnertype_id)
    except (ValueError, ValidationError) as e:
        LOG.warning('Database lookup for runnertype with id="%s" resulted in '
                    'exception: %s', runnertype_id, e)
        raise StackStormDBObjectNotFoundError('Unable to find runnertype with '
                                              'id="%s"' % runnertype_id)

    return runnertype


def get_runnertype_by_name(runnertype_name):
    """
        Get an runnertype by name.
        On error, raise ST2ObjectNotFoundError.
    """
    try:
        runnertypes = RunnerType.query(name=runnertype_name)
    except (ValueError, ValidationError) as e:
        LOG.error('Database lookup for name="%s" resulted in exception: %s',
                  runnertype_name, e)
        raise StackStormDBObjectNotFoundError('Unable to find runnertype with name="%s"'
                                              % runnertype_name)

    if not runnertypes:
        raise StackStormDBObjectNotFoundError('Unable to find RunnerType with name="%s"'
                                              % runnertype_name)

    if len(runnertypes) > 1:
        LOG.warning('More than one RunnerType returned from DB lookup by name. '
                    'Result list is: %s', runnertypes)

    return runnertypes[0]


def get_action_by_id(action_id):
    """
        Get Action by id.

        On error, raise StackStormDBObjectNotFoundError
    """
    action = None

    try:
        action = Action.get_by_id(action_id)
    except (ValueError, ValidationError) as e:
        LOG.warning('Database lookup for action with id="%s" resulted in '
                    'exception: %s', action_id, e)
        raise StackStormDBObjectNotFoundError('Unable to find action with '
                                              'id="%s"' % action_id)

    return action


def get_action_by_ref(ref):
    """
    Returns the action object from db given a string ref.

    :param ref: Reference to the trigger type db object.
    :type ref: ``str``

    :rtype action: ``object``
    """
    try:
        return Action.get_by_ref(ref)
    except ValueError as e:
        LOG.debug('Database lookup for ref="%s" resulted ' +
                  'in exception : %s.', ref, e, exc_info=True)
        return None


def get_liveaction_by_id(liveaction_id):
    """
        Get LiveAction by id.

        On error, raise ST2DBObjectNotFoundError.
    """
    liveaction = None

    try:
        liveaction = LiveAction.get_by_id(liveaction_id)
    except (ValidationError, ValueError) as e:
        LOG.error('Database lookup for LiveAction with id="%s" resulted in '
                  'exception: %s', liveaction_id, e)
        raise StackStormDBObjectNotFoundError('Unable to find LiveAction with '
                                              'id="%s"' % liveaction_id)

    return liveaction


def update_liveaction_status(status=None, result=None, context=None, end_timestamp=None,
                             liveaction_id=None, runner_info=None, liveaction_db=None,
                             publish=True):
    """
        Update the status of the specified LiveAction to the value provided in
        new_status.

        The LiveAction may be specified using either liveaction_id, or as an
        liveaction_db instance.
    """

    if (liveaction_id is None) and (liveaction_db is None):
        raise ValueError('Must specify an liveaction_id or an liveaction_db when '
                         'calling update_LiveAction_status')

    if liveaction_db is None:
        liveaction_db = get_liveaction_by_id(liveaction_id)

    if status not in LIVEACTION_STATUSES:
        raise ValueError('Attempting to set status for LiveAction "%s" '
                         'to unknown status string. Unknown status is "%s"',
                         liveaction_db, status)

    extra = {'liveaction_db': liveaction_db}
    LOG.debug('Updating ActionExection: "%s" with status="%s"', liveaction_db.id, status,
              extra=extra)

    old_status = liveaction_db.status
    liveaction_db.status = status

    if result:
        liveaction_db.result = result

    if context:
        liveaction_db.context.update(context)

    if end_timestamp:
        liveaction_db.end_timestamp = end_timestamp

    if runner_info:
        liveaction_db.runner_info = runner_info

    liveaction_db = LiveAction.add_or_update(liveaction_db)

    LOG.debug('Updated status for LiveAction object.', extra=extra)

    if publish and status != old_status:
        LiveAction.publish_status(liveaction_db)
        LOG.debug('Published status for LiveAction object.', extra=extra)

    return liveaction_db


def serialize_positional_argument(argument_type, argument_value):
    """
    Serialize the provided positional argument.

    Note: Serialization is NOT performed recursively since it doesn't make much
    sense for shell script actions (only the outter / top level value is
    serialized).
    """
    if argument_type in ['string', 'number', 'float']:
        argument_value = str(argument_value) if argument_value else ''
    elif argument_type == 'boolean':
        # Booleans are serialized as string "1" and "0"
        if argument_value is not None:
            argument_value = '1' if bool(argument_value) else '0'
        else:
            argument_value = ''
    elif argument_type == 'list':
        # Lists are serialized a comma delimited string (foo,bar,baz)
        argument_value = ','.join(argument_value) if argument_value else ''
    elif argument_type == 'object':
        # Objects are serialized as JSON
        argument_value = json.dumps(argument_value) if argument_value else ''
    elif argument_type is 'null':
        # None / null is serialized as en empty string
        argument_value = ''
    else:
        # Other values are simply cast to strings
        argument_value = str(argument_value) if argument_value else ''

    return argument_value


def get_args(action_parameters, action_db):
    """

    Get and serialize positional and named arguments.

    :return: (positional_args, named_args)
    :rtype: (``str``, ``dict``)
    """
    position_args_dict = _get_position_arg_dict(action_parameters, action_db)

    action_db_parameters = action_db.parameters or {}

    positional_args = []
    positional_args_keys = set()
    for _, arg in six.iteritems(position_args_dict):
        arg_type = action_db_parameters.get(arg, {}).get('type', None)

        # Perform serialization for positional arguments
        arg_value = action_parameters.get(arg, None)
        arg_value = serialize_positional_argument(argument_type=arg_type,
                                                  argument_value=arg_value)

        positional_args.append(arg_value)
        positional_args_keys.add(arg)

    named_args = {}
    for param in action_parameters:
        if param not in positional_args_keys:
            named_args[param] = action_parameters.get(param)

    return positional_args, named_args


def _get_position_arg_dict(action_parameters, action_db):
    action_db_params = action_db.parameters

    args_dict = {}
    for param in action_db_params:
        param_meta = action_db_params.get(param, None)
        if param_meta is not None:
            pos = param_meta.get('position')
            if pos is not None:
                args_dict[pos] = param
    args_dict = OrderedDict(sorted(args_dict.items()))
    return args_dict


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 simplejson as json
17			from collections import OrderedDict
18
19			from mongoengine import ValidationError
20			import six
21
22			from st2common import log as logging
23			from st2common.constants.action import LIVEACTION_STATUSES
24			from st2common.exceptions.db import StackStormDBObjectNotFoundError
25			from st2common.persistence.action import Action
26			from st2common.persistence.liveaction import LiveAction
27			from st2common.persistence.runner import RunnerType
28
29			LOG = logging.getLogger(__name__)
30
31
32			__all__ = [
33			'get_action_parameters_specs',
34			'get_runnertype_by_id',
35			'get_runnertype_by_name',
36			'get_action_by_id',
37			'get_action_by_ref',
38			'get_liveaction_by_id',
39			'update_liveaction_status',
40			'serialize_positional_argument',
41			'get_args'
42			]
43
44
45			def get_action_parameters_specs(action_ref):
46			"""
47			Retrieve parameters specifications schema for the provided action reference.
48
49			Note: This function returns a union of action and action runner parameters.
50
51			:param action_ref: Action reference.
52			:type action_ref: ``str``
53
54			:rtype: ``dict``
55			"""
56			action_db = get_action_by_ref(ref=action_ref)
57
58			parameters = {}
59			if not action_db:
60			return parameters
61
62			runner_type_name = action_db.runner_type['name']
63			runner_type_db = get_runnertype_by_name(runnertype_name=runner_type_name)
64
65			# Runner type parameters should be added first before the action parameters.
66			parameters.update(runner_type_db['runner_parameters'])
67			parameters.update(action_db.parameters)
68
69			return parameters
70
71
72			def get_runnertype_by_id(runnertype_id):
73			"""
74			Get RunnerType by id.
75
76			On error, raise StackStormDBObjectNotFoundError
77			"""
78			try:
79			runnertype = RunnerType.get_by_id(runnertype_id)
80			except (ValueError, ValidationError) as e:
81			LOG.warning('Database lookup for runnertype with id="%s" resulted in '
82			'exception: %s', runnertype_id, e)
83			raise StackStormDBObjectNotFoundError('Unable to find runnertype with '
84			'id="%s"' % runnertype_id)
85
86			return runnertype
87
88
89			def get_runnertype_by_name(runnertype_name):
90			"""
91			Get an runnertype by name.
92			On error, raise ST2ObjectNotFoundError.
93			"""
94			try:
95			runnertypes = RunnerType.query(name=runnertype_name)
96			except (ValueError, ValidationError) as e:
97			LOG.error('Database lookup for name="%s" resulted in exception: %s',
98			runnertype_name, e)
99			raise StackStormDBObjectNotFoundError('Unable to find runnertype with name="%s"'
100			% runnertype_name)
101
102			if not runnertypes:
103			raise StackStormDBObjectNotFoundError('Unable to find RunnerType with name="%s"'
104			% runnertype_name)
105
106			if len(runnertypes) > 1:
107			LOG.warning('More than one RunnerType returned from DB lookup by name. '
108			'Result list is: %s', runnertypes)
109
110			return runnertypes[0]
111
112
113			def get_action_by_id(action_id):
114			"""
115			Get Action by id.
116
117			On error, raise StackStormDBObjectNotFoundError
118			"""
119			action = None
120
121			try:
122			action = Action.get_by_id(action_id)
123			except (ValueError, ValidationError) as e:
124			LOG.warning('Database lookup for action with id="%s" resulted in '
125			'exception: %s', action_id, e)
126			raise StackStormDBObjectNotFoundError('Unable to find action with '
127			'id="%s"' % action_id)
128
129			return action
130
131
132			def get_action_by_ref(ref):
133			"""
134			Returns the action object from db given a string ref.
135
136			:param ref: Reference to the trigger type db object.
137			:type ref: ``str``
138
139			:rtype action: ``object``
140			"""
141			try:
142			return Action.get_by_ref(ref)
143			except ValueError as e:
144			LOG.debug('Database lookup for ref="%s" resulted ' +
145			'in exception : %s.', ref, e, exc_info=True)
146			return None
147
148
149			def get_liveaction_by_id(liveaction_id):
150			"""
151			Get LiveAction by id.
152
153			On error, raise ST2DBObjectNotFoundError.
154			"""
155			liveaction = None
156
157			try:
158			liveaction = LiveAction.get_by_id(liveaction_id)
159			except (ValidationError, ValueError) as e:
160			LOG.error('Database lookup for LiveAction with id="%s" resulted in '
161			'exception: %s', liveaction_id, e)
162			raise StackStormDBObjectNotFoundError('Unable to find LiveAction with '
163			'id="%s"' % liveaction_id)
164
165			return liveaction
166
167
168			def update_liveaction_status(status=None, result=None, context=None, end_timestamp=None,
169			liveaction_id=None, runner_info=None, liveaction_db=None,
170			publish=True):
171			"""
172			Update the status of the specified LiveAction to the value provided in
173			new_status.
174
175			The LiveAction may be specified using either liveaction_id, or as an
176			liveaction_db instance.
177			"""
178
179			if (liveaction_id is None) and (liveaction_db is None):
180			raise ValueError('Must specify an liveaction_id or an liveaction_db when '
181			'calling update_LiveAction_status')
182
183			if liveaction_db is None:
184			liveaction_db = get_liveaction_by_id(liveaction_id)
185
186			if status not in LIVEACTION_STATUSES:
187			raise ValueError('Attempting to set status for LiveAction "%s" '
188			'to unknown status string. Unknown status is "%s"',
189			liveaction_db, status)
190
191			extra = {'liveaction_db': liveaction_db}
192			LOG.debug('Updating ActionExection: "%s" with status="%s"', liveaction_db.id, status,
193			extra=extra)
194
195			old_status = liveaction_db.status
196			liveaction_db.status = status
197
198			if result:
199			liveaction_db.result = result
200
201			if context:
202			liveaction_db.context.update(context)
203
204			if end_timestamp:
205			liveaction_db.end_timestamp = end_timestamp
206
207			if runner_info:
208			liveaction_db.runner_info = runner_info
209
210			liveaction_db = LiveAction.add_or_update(liveaction_db)
211
212			LOG.debug('Updated status for LiveAction object.', extra=extra)
213
214			if publish and status != old_status:
215			LiveAction.publish_status(liveaction_db)
216			LOG.debug('Published status for LiveAction object.', extra=extra)
217
218			return liveaction_db
219
220
221			def serialize_positional_argument(argument_type, argument_value):
222			"""
223			Serialize the provided positional argument.
224
225			Note: Serialization is NOT performed recursively since it doesn't make much
226			sense for shell script actions (only the outter / top level value is
227			serialized).
228			"""
229			if argument_type in ['string', 'number', 'float']:
230			argument_value = str(argument_value) if argument_value else ''
231			elif argument_type == 'boolean':
232			# Booleans are serialized as string "1" and "0"
233			if argument_value is not None:
234			argument_value = '1' if bool(argument_value) else '0'
235			else:
236			argument_value = ''
237			elif argument_type == 'list':
238			# Lists are serialized a comma delimited string (foo,bar,baz)
239			argument_value = ','.join(argument_value) if argument_value else ''
240			elif argument_type == 'object':
241			# Objects are serialized as JSON
242			argument_value = json.dumps(argument_value) if argument_value else ''
243			elif argument_type is 'null':
244			# None / null is serialized as en empty string
245			argument_value = ''
246			else:
247			# Other values are simply cast to strings
248			argument_value = str(argument_value) if argument_value else ''
249
250			return argument_value
251
252
253			def get_args(action_parameters, action_db):
254			"""
255
256			Get and serialize positional and named arguments.
257
258			:return: (positional_args, named_args)
259			:rtype: (``str``, ``dict``)
260			"""
261			position_args_dict = _get_position_arg_dict(action_parameters, action_db)
262
263			action_db_parameters = action_db.parameters or {}
264
265			positional_args = []
266			positional_args_keys = set()
267			for _, arg in six.iteritems(position_args_dict):
268			arg_type = action_db_parameters.get(arg, {}).get('type', None)
269
270			# Perform serialization for positional arguments
271			arg_value = action_parameters.get(arg, None)
272			arg_value = serialize_positional_argument(argument_type=arg_type,
273			argument_value=arg_value)
274
275			positional_args.append(arg_value)
276			positional_args_keys.add(arg)
277
278			named_args = {}
279			for param in action_parameters:
280			if param not in positional_args_keys:
281			named_args[param] = action_parameters.get(param)
282
283			return positional_args, named_args
284
285
286			def _get_position_arg_dict(action_parameters, action_db):
287			action_db_params = action_db.parameters
288
289			args_dict = {}
290			for param in action_db_params:
291			param_meta = action_db_params.get(param, None)
292			if param_meta is not None:
293			pos = param_meta.get('position')
294			if pos is not None:
295			args_dict[pos] = param
296			args_dict = OrderedDict(sorted(args_dict.items()))
297			return args_dict
298

StackStorm / st2

Push — develop ( 1df8f4...7d872c )

st2common.util.serialize_positional_argument() F

Complexity

Size

Duplication

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like