st2actions.runners.PythonActionWrapper.__init__() - Code Metrics - Inspection of "remove unused logging imports and globals" - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop-v1.3.1 ( acb9c3...7a20bf )

unknown

created 2016-02-04 19:24 UTC

st2actions.runners.PythonActionWrapper.init() B

↳ Parent: st2actions.runners.PythonActionWrapper

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	27
rs	8.8571
cc	2

# 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 sys
import json
import argparse
import logging as stdlib_logging

from st2common import log as logging
from st2actions import config
from st2actions.runners.pythonrunner import Action
from st2common.util import loader as action_loader
from st2common.util.config_parser import ContentPackConfigParser
from st2common.constants.action import ACTION_OUTPUT_RESULT_DELIMITER
from st2common.service_setup import db_setup
from st2common.services.datastore import DatastoreService

__all__ = [
    'PythonActionWrapper'
]

LOG = logging.getLogger(__name__)


class PythonActionWrapper(object):
    def __init__(self, pack, file_path, parameters=None, parent_args=None):
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        """
        :param pack: Name of the pack this action belongs to.
        :type pack: ``str``

        :param file_path: Path to the action module.
        :type file_path: ``str``

        :param parameters: action parameters.
        :type parameters: ``dict`` or ``None``

        :param parent_args: Command line arguments passed to the parent process.
        :type parse_args: ``list``
        """
        db_setup()

        self._pack = pack
        self._file_path = file_path
        self._parameters = parameters or {}
        self._parent_args = parent_args or []
        self._class_name = None
        self._logger = logging.getLogger('PythonActionWrapper')

        try:
            config.parse_args(args=self._parent_args)
        except Exception:
            pass

    def run(self):
        action = self._get_action_instance()
        output = action.run(**self._parameters)

        # Print output to stdout so the parent can capture it
        sys.stdout.write(ACTION_OUTPUT_RESULT_DELIMITER)
        print_output = None
        try:
            print_output = json.dumps(output)
        except:
            print_output = str(output)
        sys.stdout.write(print_output + '\n')
        sys.stdout.write(ACTION_OUTPUT_RESULT_DELIMITER)

    def _get_action_instance(self):
        actions_cls = action_loader.register_plugin(Action, self._file_path)
        action_cls = actions_cls[0] if actions_cls and len(actions_cls) > 0 else None

        if not action_cls:
            raise Exception('File "%s" has no action or the file doesn\'t exist.' %
                            (self._file_path))

        config_parser = ContentPackConfigParser(pack_name=self._pack)
        config = config_parser.get_action_config(action_file_path=self._file_path)
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param

        if config:
            LOG.info('Using config "%s" for action "%s"' % (config.file_path,
                                                            self._file_path))

            action_instance = action_cls(config=config.config)
        else:
            LOG.info('No config found for action "%s"' % (self._file_path))
            action_instance = action_cls(config={})

        # Setup action_instance proeprties
        action_instance.logger = self._set_up_logger(action_cls.__name__)
        action_instance.datastore = DatastoreService(logger=action_instance.logger,
                                                     pack_name=self._pack,
                                                     class_name=action_cls.__name__,
                                                     api_username="action_service")

        return action_instance

    def _set_up_logger(self, action_name):
        """
        Set up a logger which logs all the messages with level DEBUG
        and above to stderr.
        """
        logger_name = 'actions.python.%s' % (action_name)
        logger = logging.getLogger(logger_name)

        console = stdlib_logging.StreamHandler()
        console.setLevel(stdlib_logging.DEBUG)

        formatter = stdlib_logging.Formatter('%(name)-12s: %(levelname)-8s %(message)s')
        console.setFormatter(formatter)
        logger.addHandler(console)
        logger.setLevel(stdlib_logging.DEBUG)

        return logger


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Python action runner process wrapper')
    parser.add_argument('--pack', required=True,
                        help='Name of the pack this action belongs to')
    parser.add_argument('--file-path', required=True,
                        help='Path to the action module')
    parser.add_argument('--parameters', required=False,
                        help='Serialized action parameters')
    parser.add_argument('--parent-args', required=False,
                        help='Command line arguments passed to the parent process')
    args = parser.parse_args()

    parameters = args.parameters
    parameters = json.loads(parameters) if parameters else {}
    parent_args = json.loads(args.parent_args) if args.parent_args else []
    assert isinstance(parent_args, list)

    obj = PythonActionWrapper(pack=args.pack,
                              file_path=args.file_path,
                              parameters=parameters,
                              parent_args=parent_args)

    obj.run()


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 sys
17			import json
18			import argparse
19			import logging as stdlib_logging
20
21			from st2common import log as logging
22			from st2actions import config
23			from st2actions.runners.pythonrunner import Action
24			from st2common.util import loader as action_loader
25			from st2common.util.config_parser import ContentPackConfigParser
26			from st2common.constants.action import ACTION_OUTPUT_RESULT_DELIMITER
27			from st2common.service_setup import db_setup
28			from st2common.services.datastore import DatastoreService
29
30			__all__ = [
31			'PythonActionWrapper'
32			]
33
34			LOG = logging.getLogger(__name__)
35
36
37			class PythonActionWrapper(object):
38			def __init__(self, pack, file_path, parameters=None, parent_args=None):
			0 ignored issues – show Comprehensibility Bug introduced 2016-02-04 18:21 UTC by Report Bug Copy Issue Report `parent_args` is re-defining a name which is already available in the outer-scope (previously defined on line `142`). It is generally a bad practice to shadow variables from the outer-scope. In most cases, this is done unintentionally and might lead to unexpected behavior: param = 5 class Foo: def __init__(self, param): # "param" would be flagged here self.param = param Loading history... Comprehensibility Bug introduced 2016-02-04 18:21 UTC by Report Bug Copy Issue Report `parameters` is re-defining a name which is already available in the outer-scope (previously defined on line `140`). It is generally a bad practice to shadow variables from the outer-scope. In most cases, this is done unintentionally and might lead to unexpected behavior: param = 5 class Foo: def __init__(self, param): # "param" would be flagged here self.param = param Loading history...
39			"""
40			:param pack: Name of the pack this action belongs to.
41			:type pack: ``str``
42
43			:param file_path: Path to the action module.
44			:type file_path: ``str``
45
46			:param parameters: action parameters.
47			:type parameters: ``dict`` or ``None``
48
49			:param parent_args: Command line arguments passed to the parent process.
50			:type parse_args: ``list``
51			"""
52			db_setup()
53
54			self._pack = pack
55			self._file_path = file_path
56			self._parameters = parameters or {}
57			self._parent_args = parent_args or []
58			self._class_name = None
59			self._logger = logging.getLogger('PythonActionWrapper')
60
61			try:
62			config.parse_args(args=self._parent_args)
63			except Exception:
64			pass
65
66			def run(self):
67			action = self._get_action_instance()
68			output = action.run(**self._parameters)
69
70			# Print output to stdout so the parent can capture it
71			sys.stdout.write(ACTION_OUTPUT_RESULT_DELIMITER)
72			print_output = None
73			try:
74			print_output = json.dumps(output)
75			except:
76			print_output = str(output)
77			sys.stdout.write(print_output + '\n')
78			sys.stdout.write(ACTION_OUTPUT_RESULT_DELIMITER)
79
80			def _get_action_instance(self):
81			actions_cls = action_loader.register_plugin(Action, self._file_path)
82			action_cls = actions_cls[0] if actions_cls and len(actions_cls) > 0 else None
83
84			if not action_cls:
85			raise Exception('File "%s" has no action or the file doesn\'t exist.' %
86			(self._file_path))
87
88			config_parser = ContentPackConfigParser(pack_name=self._pack)
89			config = config_parser.get_action_config(action_file_path=self._file_path)
			0 ignored issues – show Comprehensibility Bug introduced 2016-02-04 18:21 UTC by Report Bug Copy Issue Report `config` is re-defining a name which is already available in the outer-scope (previously defined on line `22`). It is generally a bad practice to shadow variables from the outer-scope. In most cases, this is done unintentionally and might lead to unexpected behavior: param = 5 class Foo: def __init__(self, param): # "param" would be flagged here self.param = param Loading history...
90
91			if config:
92			LOG.info('Using config "%s" for action "%s"' % (config.file_path,
93			self._file_path))
94
95			action_instance = action_cls(config=config.config)
96			else:
97			LOG.info('No config found for action "%s"' % (self._file_path))
98			action_instance = action_cls(config={})
99
100			# Setup action_instance proeprties
101			action_instance.logger = self._set_up_logger(action_cls.__name__)
102			action_instance.datastore = DatastoreService(logger=action_instance.logger,
103			pack_name=self._pack,
104			class_name=action_cls.__name__,
105			api_username="action_service")
106
107			return action_instance
108
109			def _set_up_logger(self, action_name):
110			"""
111			Set up a logger which logs all the messages with level DEBUG
112			and above to stderr.
113			"""
114			logger_name = 'actions.python.%s' % (action_name)
115			logger = logging.getLogger(logger_name)
116
117			console = stdlib_logging.StreamHandler()
118			console.setLevel(stdlib_logging.DEBUG)
119
120			formatter = stdlib_logging.Formatter('%(name)-12s: %(levelname)-8s %(message)s')
121			console.setFormatter(formatter)
122			logger.addHandler(console)
123			logger.setLevel(stdlib_logging.DEBUG)
124
125			return logger
126
127
128			if __name__ == '__main__':
129			parser = argparse.ArgumentParser(description='Python action runner process wrapper')
130			parser.add_argument('--pack', required=True,
131			help='Name of the pack this action belongs to')
132			parser.add_argument('--file-path', required=True,
133			help='Path to the action module')
134			parser.add_argument('--parameters', required=False,
135			help='Serialized action parameters')
136			parser.add_argument('--parent-args', required=False,
137			help='Command line arguments passed to the parent process')
138			args = parser.parse_args()
139
140			parameters = args.parameters
141			parameters = json.loads(parameters) if parameters else {}
142			parent_args = json.loads(args.parent_args) if args.parent_args else []
143			assert isinstance(parent_args, list)
144
145			obj = PythonActionWrapper(pack=args.pack,
146			file_path=args.file_path,
147			parameters=parameters,
148			parent_args=parent_args)
149
150			obj.run()
151

Plexxi / st2

GitHub Access Token became invalid

Push — develop-v1.3.1 ( acb9c3...7a20bf )

st2actions.runners.PythonActionWrapper.__init__() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like

st2actions.runners.PythonActionWrapper.init() B