PollingAsyncActionRunner.is_polling_enabled() - Code Metrics - Inspection of "Refactor mistral runner to support callback" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#4000)

by W

created 2018-02-27 01:15 UTC

PollingAsyncActionRunner.is_polling_enabled() A

↳ Parent: PollingAsyncActionRunner

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
dl	0
loc	3
rs	10
c	1
b	0
f	0

# 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.

from __future__ import absolute_import
import abc

import six
import yaml
from oslo_config import cfg

from st2common import log as logging
from st2common.constants import action as action_constants
from st2common.constants import pack as pack_constants
from st2common.exceptions.actionrunner import ActionRunnerCreateError
from st2common.util import action_db as action_utils
from st2common.util.loader import register_runner
from st2common.util.loader import register_callback_module
from st2common.util.api import get_full_public_api_url
from st2common.util.deprecation import deprecated

__all__ = [
    'ActionRunner',
    'AsyncActionRunner',
    'PollingAsyncActionRunner',
    'ShellRunnerMixin',

    'get_runner',
    'get_metadata'
]


LOG = logging.getLogger(__name__)

# constants to lookup in runner_parameters
RUNNER_COMMAND = 'cmd'


def get_runner(package_name, module_name, config=None):
    """
    Load the module and return an instance of the runner.
    """

    if not package_name:
        # Backward compatibility for Pre 2.7.0 where package name always equaled module name
        package_name = module_name

    LOG.debug('Runner loading Python module: %s.%s', package_name, module_name)

    try:
        # TODO: Explore modifying this to support register_plugin
        module = register_runner(package_name=package_name, module_name=module_name)
    except Exception as e:
        msg = ('Failed to import runner module %s.%s' % (package_name, module_name))
        LOG.exception(msg)

        raise ActionRunnerCreateError('%s\n\n%s' % (msg, str(e)))

    LOG.debug('Instance of runner module: %s', module)

    if config:
        runner_kwargs = {'config': config}
    else:
        runner_kwargs = {}

    runner = module.get_runner(**runner_kwargs)
    LOG.debug('Instance of runner: %s', runner)
    return runner


def get_metadata(package_name):
    """
    Return runner related metadata for the provided runner package name.

    :rtype: ``list`` of ``dict``
    """
    import pkg_resources

    file_path = pkg_resources.resource_filename(package_name, 'runner.yaml')

    with open(file_path, 'r') as fp:
        content = fp.read()

    metadata = yaml.safe_load(content)
    return metadata


@six.add_metaclass(abc.ABCMeta)
class ActionRunner(object):
    """
        The interface that must be implemented by each StackStorm
        Action Runner implementation.
    """

    def __init__(self, runner_id):
        """
        :param id: Runner id.
        :type id: ``str``
        """
        self.runner_id = runner_id

        self.runner_type_db = None
        self.runner_parameters = None
        self.action = None
        self.action_name = None
        self.liveaction = None
        self.liveaction_id = None
        self.execution = None
        self.execution_id = None
        self.entry_point = None
        self.libs_dir_path = None
        self.context = None
        self.callback = None
        self.auth_token = None
        self.rerun_ex_ref = None

    def pre_run(self):
        runner_enabled = getattr(self.runner_type_db, 'enabled', True)
        runner_name = getattr(self.runner_type_db, 'name', 'unknown')
        if not runner_enabled:
            msg = ('Runner "%s" has been disabled by the administrator' %
                   (runner_name))
            raise ValueError(msg)

    # Run will need to take an action argument
    # Run may need result data argument
    @abc.abstractmethod
    def run(self, action_parameters):
        raise NotImplementedError()

    def pause(self):
        runner_name = getattr(self.runner_type_db, 'name', 'unknown')
        raise NotImplementedError('Pause is not supported for runner %s.' % runner_name)

    def resume(self):
        runner_name = getattr(self.runner_type_db, 'name', 'unknown')
        raise NotImplementedError('Resume is not supported for runner %s.' % runner_name)

    def cancel(self):
        return (
            action_constants.LIVEACTION_STATUS_CANCELED,
            self.liveaction.result,
            self.liveaction.context
        )

    def post_run(self, status, result):
        callback = self.callback or {}

        if callback and not (set(['url', 'source']) - set(callback.keys())):

            callback_url = callback['url']
            callback_module_name = callback['source']

            try:
                callback_module = register_callback_module(callback_module_name)
            except:
                LOG.exception('Failed importing callback module: %s', callback_module_name)

            callback_handler = callback_module.get_instance()

            callback_handler.callback(
                callback_url,
                self.context,
                status,
                result
            )

    @deprecated
    def get_pack_name(self):
        return self.get_pack_ref()

    def get_pack_ref(self):
        """
        Retrieve pack name for the action which is being currently executed.

        :rtype: ``str``
        """
        if self.action:
            return self.action.pack

        return pack_constants.DEFAULT_PACK_NAME

    def get_user(self):
        """
        Retrieve a name of the user which triggered this action execution.

        :rtype: ``str``
        """
        context = getattr(self, 'context', {}) or {}
        user = context.get('user', cfg.CONF.system_user.user)

        return user

    def _get_common_action_env_variables(self):
        """
        Retrieve common ST2_ACTION_ environment variables which will be available to the action.

        Note: Environment variables are prefixed with ST2_ACTION_* so they don't clash with CLI
        environment variables.

        :rtype: ``dict``
        """
        result = {}
        result['ST2_ACTION_PACK_NAME'] = self.get_pack_ref()
        result['ST2_ACTION_EXECUTION_ID'] = str(self.execution_id)
        result['ST2_ACTION_API_URL'] = get_full_public_api_url()

        if self.auth_token:
            result['ST2_ACTION_AUTH_TOKEN'] = self.auth_token.token

        return result

    def __str__(self):
        attrs = ', '.join(['%s=%s' % (k, v) for k, v in six.iteritems(self.__dict__)])
        return '%s@%s(%s)' % (self.__class__.__name__, str(id(self)), attrs)


@six.add_metaclass(abc.ABCMeta)
class AsyncActionRunner(ActionRunner):
    pass


@six.add_metaclass(abc.ABCMeta)
class PollingAsyncActionRunner(AsyncActionRunner):

    @classmethod
    def is_polling_enabled(cls):
        return True


class ShellRunnerMixin(object):
    """
    Class which contains utility functions to be used by shell runners.
    """

    def _transform_named_args(self, named_args):
        """
        Transform named arguments to the final form.

        :param named_args: Named arguments.
        :type named_args: ``dict``

        :rtype: ``dict``
        """
        if named_args:
            return {self._kwarg_op + k: v for (k, v) in six.iteritems(named_args)}
        return None

    def _get_script_args(self, action_parameters):
        """
        :param action_parameters: Action parameters.
        :type action_parameters: ``dict``

        :return: (positional_args, named_args)
        :rtype: (``str``, ``dict``)
        """
        # TODO: return list for positional args, command classes should escape it
        # and convert it to string

        is_script_run_as_cmd = self.runner_parameters.get(RUNNER_COMMAND, None)

        pos_args = ''
        named_args = {}

        if is_script_run_as_cmd:
            pos_args = self.runner_parameters.get(RUNNER_COMMAND, '')
            named_args = action_parameters
        else:
            pos_args, named_args = action_utils.get_args(action_parameters, self.action)

        return pos_args, named_args


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			from __future__ import absolute_import
17			import abc
18
19			import six
20			import yaml
21			from oslo_config import cfg
22
23			from st2common import log as logging
24			from st2common.constants import action as action_constants
25			from st2common.constants import pack as pack_constants
26			from st2common.exceptions.actionrunner import ActionRunnerCreateError
27			from st2common.util import action_db as action_utils
28			from st2common.util.loader import register_runner
29			from st2common.util.loader import register_callback_module
30			from st2common.util.api import get_full_public_api_url
31			from st2common.util.deprecation import deprecated
32
33			__all__ = [
34			'ActionRunner',
35			'AsyncActionRunner',
36			'PollingAsyncActionRunner',
37			'ShellRunnerMixin',
38
39			'get_runner',
40			'get_metadata'
41			]
42
43
44			LOG = logging.getLogger(__name__)
45
46			# constants to lookup in runner_parameters
47			RUNNER_COMMAND = 'cmd'
48
49
50			def get_runner(package_name, module_name, config=None):
51			"""
52			Load the module and return an instance of the runner.
53			"""
54
55			if not package_name:
56			# Backward compatibility for Pre 2.7.0 where package name always equaled module name
57			package_name = module_name
58
59			LOG.debug('Runner loading Python module: %s.%s', package_name, module_name)
60
61			try:
62			# TODO: Explore modifying this to support register_plugin
63			module = register_runner(package_name=package_name, module_name=module_name)
64			except Exception as e:
65			msg = ('Failed to import runner module %s.%s' % (package_name, module_name))
66			LOG.exception(msg)
67
68			raise ActionRunnerCreateError('%s\n\n%s' % (msg, str(e)))
69
70			LOG.debug('Instance of runner module: %s', module)
71
72			if config:
73			runner_kwargs = {'config': config}
74			else:
75			runner_kwargs = {}
76
77			runner = module.get_runner(**runner_kwargs)
78			LOG.debug('Instance of runner: %s', runner)
79			return runner
80
81
82			def get_metadata(package_name):
83			"""
84			Return runner related metadata for the provided runner package name.
85
86			:rtype: ``list`` of ``dict``
87			"""
88			import pkg_resources
89
90			file_path = pkg_resources.resource_filename(package_name, 'runner.yaml')
91
92			with open(file_path, 'r') as fp:
93			content = fp.read()
94
95			metadata = yaml.safe_load(content)
96			return metadata
97
98
99			@six.add_metaclass(abc.ABCMeta)
100			class ActionRunner(object):
101			"""
102			The interface that must be implemented by each StackStorm
103			Action Runner implementation.
104			"""
105
106			def __init__(self, runner_id):
107			"""
108			:param id: Runner id.
109			:type id: ``str``
110			"""
111			self.runner_id = runner_id
112
113			self.runner_type_db = None
114			self.runner_parameters = None
115			self.action = None
116			self.action_name = None
117			self.liveaction = None
118			self.liveaction_id = None
119			self.execution = None
120			self.execution_id = None
121			self.entry_point = None
122			self.libs_dir_path = None
123			self.context = None
124			self.callback = None
125			self.auth_token = None
126			self.rerun_ex_ref = None
127
128			def pre_run(self):
129			runner_enabled = getattr(self.runner_type_db, 'enabled', True)
130			runner_name = getattr(self.runner_type_db, 'name', 'unknown')
131			if not runner_enabled:
132			msg = ('Runner "%s" has been disabled by the administrator' %
133			(runner_name))
134			raise ValueError(msg)
135
136			# Run will need to take an action argument
137			# Run may need result data argument
138			@abc.abstractmethod
139			def run(self, action_parameters):
140			raise NotImplementedError()
141
142			def pause(self):
143			runner_name = getattr(self.runner_type_db, 'name', 'unknown')
144			raise NotImplementedError('Pause is not supported for runner %s.' % runner_name)
145
146			def resume(self):
147			runner_name = getattr(self.runner_type_db, 'name', 'unknown')
148			raise NotImplementedError('Resume is not supported for runner %s.' % runner_name)
149
150			def cancel(self):
151			return (
152			action_constants.LIVEACTION_STATUS_CANCELED,
153			self.liveaction.result,
154			self.liveaction.context
155			)
156
157			def post_run(self, status, result):
158			callback = self.callback or {}
159
160			if callback and not (set(['url', 'source']) - set(callback.keys())):
			0 ignored issues – show Unused Code Coding Style introduced 2017-04-18 08:10 UTC by Report Bug Copy Issue Report There is an unnecessary parenthesis after `not`. Loading history...
161			callback_url = callback['url']
162			callback_module_name = callback['source']
163
164			try:
165			callback_module = register_callback_module(callback_module_name)
166			except:
167			LOG.exception('Failed importing callback module: %s', callback_module_name)
168
169			callback_handler = callback_module.get_instance()
170
171			callback_handler.callback(
172			callback_url,
173			self.context,
174			status,
175			result
176			)
177
178			@deprecated
179			def get_pack_name(self):
180			return self.get_pack_ref()
181
182			def get_pack_ref(self):
183			"""
184			Retrieve pack name for the action which is being currently executed.
185
186			:rtype: ``str``
187			"""
188			if self.action:
189			return self.action.pack
190
191			return pack_constants.DEFAULT_PACK_NAME
192
193			def get_user(self):
194			"""
195			Retrieve a name of the user which triggered this action execution.
196
197			:rtype: ``str``
198			"""
199			context = getattr(self, 'context', {}) or {}
200			user = context.get('user', cfg.CONF.system_user.user)
201
202			return user
203
204			def _get_common_action_env_variables(self):
205			"""
206			Retrieve common ST2_ACTION_ environment variables which will be available to the action.
207
208			Note: Environment variables are prefixed with ST2_ACTION_* so they don't clash with CLI
209			environment variables.
210
211			:rtype: ``dict``
212			"""
213			result = {}
214			result['ST2_ACTION_PACK_NAME'] = self.get_pack_ref()
215			result['ST2_ACTION_EXECUTION_ID'] = str(self.execution_id)
216			result['ST2_ACTION_API_URL'] = get_full_public_api_url()
217
218			if self.auth_token:
219			result['ST2_ACTION_AUTH_TOKEN'] = self.auth_token.token
220
221			return result
222
223			def __str__(self):
224			attrs = ', '.join(['%s=%s' % (k, v) for k, v in six.iteritems(self.__dict__)])
225			return '%s@%s(%s)' % (self.__class__.__name__, str(id(self)), attrs)
226
227
228			@six.add_metaclass(abc.ABCMeta)
229			class AsyncActionRunner(ActionRunner):
230			pass
231
232
233			@six.add_metaclass(abc.ABCMeta)
234			class PollingAsyncActionRunner(AsyncActionRunner):
235
236			@classmethod
237			def is_polling_enabled(cls):
238			return True
239
240
241			class ShellRunnerMixin(object):
242			"""
243			Class which contains utility functions to be used by shell runners.
244			"""
245
246			def _transform_named_args(self, named_args):
247			"""
248			Transform named arguments to the final form.
249
250			:param named_args: Named arguments.
251			:type named_args: ``dict``
252
253			:rtype: ``dict``
254			"""
255			if named_args:
256			return {self._kwarg_op + k: v for (k, v) in six.iteritems(named_args)}
257			return None
258
259			def _get_script_args(self, action_parameters):
260			"""
261			:param action_parameters: Action parameters.
262			:type action_parameters: ``dict``
263
264			:return: (positional_args, named_args)
265			:rtype: (``str``, ``dict``)
266			"""
267			# TODO: return list for positional args, command classes should escape it
268			# and convert it to string
269
270			is_script_run_as_cmd = self.runner_parameters.get(RUNNER_COMMAND, None)
271
272			pos_args = ''
273			named_args = {}
274
275			if is_script_run_as_cmd:
276			pos_args = self.runner_parameters.get(RUNNER_COMMAND, '')
277			named_args = action_parameters
278			else:
279			pos_args, named_args = action_utils.get_args(action_parameters, self.action)
280
281			return pos_args, named_args
282

StackStorm / st2

Pull Request — master (#4000)

PollingAsyncActionRunner.is_polling_enabled() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like