ActionAliasExecuteCommand.__init__() - Code Metrics - Inspection of "Refactor chatops to process input alias matching s..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2895)

by Anthony

created 2016-09-14 05:21 UTC

ActionAliasExecuteCommand.init() A

↳ Parent: ActionAliasExecuteCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	1

Metric	Value
cc	1
c	2
b	0
f	1
dl	0
loc	20
rs	9.4285

# 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 st2common.exceptions.actionalias import ActionAliasAmbiguityException
from st2common.models.utils.action_alias_utils import extract_parameters_for_action_alias_db
from st2common.util.actionalias_matching import match_command_to_alias

from st2client import models
from st2client.models.action_alias import ActionAlias
from st2client.models.action import LiveAction

from st2client.commands import resource
from st2client.formatters import table

__all__ = [
    'ActionAliasBranch'
]


class ActionAliasBranch(resource.ResourceBranch):
    def __init__(self, description, app, subparsers, parent_parser=None):
        super(ActionAliasBranch, self).__init__(
            ActionAlias, description, app, subparsers,
            parent_parser=parent_parser, read_only=False,
            commands={
                'list': ActionAliasListCommand,
                'get': ActionAliasGetCommand
            })

        self.commands['match'] = ActionAliasMatchCommand(
            self.resource, self.app, self.subparsers,
            add_help=False)
        self.commands['execute'] = ActionAliasExecuteCommand(
            LiveAction, self.app, self.subparsers,
            add_help=False)


class ActionAliasListCommand(resource.ContentPackResourceListCommand):
    display_attributes = ['ref', 'pack', 'description', 'enabled']


class ActionAliasGetCommand(resource.ContentPackResourceGetCommand):
    display_attributes = ['all']
    attribute_display_order = ['id', 'ref', 'pack', 'name', 'description',
                               'enabled', 'action_ref', 'formats']


class ActionAliasMatchCommand(resource.ResourceCommand):
    display_attributes = ['id', 'name', 'description']

    def __init__(self, resource, *args, **kwargs):
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(ActionAliasMatchCommand, self).__init__(
            resource, 'match',
            'Get the list of %s that match the command text.' %
            resource.get_plural_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('match_text',
                                 metavar='command',
                                 help=help)

        self.parser.add_argument('-a', '--attr', nargs='+',
                                 default=self.display_attributes,
                                 help=('List of attributes to include in the '
                                       'output. "all" will return all '
                                       'attributes.'))
        self.parser.add_argument('-w', '--width', nargs='+', type=int,
                                 default=None,
                                 help=('Set the width of columns in output.'))

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        aliases = self.manager.get_all(**kwargs)
        matches = match_command_to_alias(args.match_text, aliases)
        return [match[0] for match in matches]  # show only alias objects

    def run_and_print(self, args, **kwargs):
        instances = self.run(args, **kwargs)
        self.print_output(instances, table.MultiColumnTable,
                          attributes=args.attr, widths=args.width,
                          json=args.json, yaml=args.yaml)


class ActionAliasExecuteCommand(resource.ResourceCommand):
    display_attributes = ['id', 'name', 'description']

    def __init__(self, resource, *args, **kwargs):
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(ActionAliasExecuteCommand, self).__init__(
            resource, 'execute',
            'Execute the command text by finding a matching ActionAlias.',
            *args, **kwargs)

        self.parser.add_argument('command_text',
                                 metavar='command',
                                 help=help)
        self.parser.add_argument('--trace-tag', '--trace_tag',
                                 help='A trace tag string to track execution later.',
                                 dest='trace_tag', required=False)
        self.parser.add_argument('--trace-id',
                                 help='Existing trace id for this execution.',
                                 dest='trace_id', required=False)
        self.parser.add_argument('-a', '--async',
                                 action='store_true', dest='async',
                                 help='Do not wait for action to finish.')
        self.parser.add_argument('-u', '--user', type=str, default=None,
                                 help='User under which to run the action (admins only).')

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        aliases = self.manager.get_all(**kwargs)
        matches = match_command_to_alias(args.command_text, aliases)
        if len(matches) > 1:
            raise ActionAliasAmbiguityException("Too many matches for provided command",
                                                matches=matches)
        elif len(matches) == 0:
            raise ActionAliasAmbiguityException("Could not locate an ActionAlias with a "
                                                "matching format command", matches=matches)
        match = matches[0]
        action_alias_db = match[0]

        if not action_alias_db.enabled:
            raise ValueError('Action alias with name "%s" is disabled.' %
                             (action_alias_db.ref))

        execution_parameters = extract_parameters_for_action_alias_db(
            action_alias_db=action_alias_db,
            format_str=matches[2],
            param_stream=args.command_text)

        execution = models.LiveAction()
        execution.action = action_alias_db.action_ref
        execution.parameters = execution_parameters
        execution.user = args.user

        if not args.trace_id and args.trace_tag:
            execution.context = {'trace_context': {'trace_tag': args.trace_tag}}

        if args.trace_id:
            execution.context = {'trace_context': {'id_': args.trace_id}}

        action_exec_mgr = self.app.client.managers['LiveAction']

        execution = action_exec_mgr.create(execution, **kwargs)
        execution = self._get_execution_result(execution=execution,
                                               action_exec_mgr=action_exec_mgr,
                                               args=args, **kwargs)
        return execution

    def run_and_print(self, args, **kwargs):
        instances = self.run(args, **kwargs)
        self.print_output(instances, table.MultiColumnTable,
                          attributes=args.attr, widths=args.width,
                          json=args.json, yaml=args.yaml)


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 st2common.exceptions.actionalias import ActionAliasAmbiguityException
17			from st2common.models.utils.action_alias_utils import extract_parameters_for_action_alias_db
18			from st2common.util.actionalias_matching import match_command_to_alias
19
20			from st2client import models
21			from st2client.models.action_alias import ActionAlias
22			from st2client.models.action import LiveAction
23
24			from st2client.commands import resource
25			from st2client.formatters import table
26
27			__all__ = [
28			'ActionAliasBranch'
29			]
30
31
32			class ActionAliasBranch(resource.ResourceBranch):
33			def __init__(self, description, app, subparsers, parent_parser=None):
34			super(ActionAliasBranch, self).__init__(
35			ActionAlias, description, app, subparsers,
36			parent_parser=parent_parser, read_only=False,
37			commands={
38			'list': ActionAliasListCommand,
39			'get': ActionAliasGetCommand
40			})
41
42			self.commands['match'] = ActionAliasMatchCommand(
43			self.resource, self.app, self.subparsers,
44			add_help=False)
45			self.commands['execute'] = ActionAliasExecuteCommand(
46			LiveAction, self.app, self.subparsers,
47			add_help=False)
48
49
50			class ActionAliasListCommand(resource.ContentPackResourceListCommand):
51			display_attributes = ['ref', 'pack', 'description', 'enabled']
52
53
54			class ActionAliasGetCommand(resource.ContentPackResourceGetCommand):
55			display_attributes = ['all']
56			attribute_display_order = ['id', 'ref', 'pack', 'name', 'description',
57			'enabled', 'action_ref', 'formats']
58
59
60			class ActionAliasMatchCommand(resource.ResourceCommand):
61			display_attributes = ['id', 'name', 'description']
62
63			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-09-14 04:30 UTC by Report Bug Copy Issue Report `resource` is re-defining a name which is already available in the outer-scope (previously defined on line `24`). 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...
64			super(ActionAliasMatchCommand, self).__init__(
65			resource, 'match',
66			'Get the list of %s that match the command text.' %
67			resource.get_plural_display_name().lower(),
68			args, *kwargs)
69
70			self.parser.add_argument('match_text',
71			metavar='command',
72			help=help)
73
74			self.parser.add_argument('-a', '--attr', nargs='+',
75			default=self.display_attributes,
76			help=('List of attributes to include in the '
77			'output. "all" will return all '
78			'attributes.'))
79			self.parser.add_argument('-w', '--width', nargs='+', type=int,
80			default=None,
81			help=('Set the width of columns in output.'))
82
83			@resource.add_auth_token_to_kwargs_from_cli
84			def run(self, args, **kwargs):
85			aliases = self.manager.get_all(**kwargs)
86			matches = match_command_to_alias(args.match_text, aliases)
87			return [match[0] for match in matches] # show only alias objects
88
89			def run_and_print(self, args, **kwargs):
90			instances = self.run(args, **kwargs)
91			self.print_output(instances, table.MultiColumnTable,
92			attributes=args.attr, widths=args.width,
93			json=args.json, yaml=args.yaml)
94
95
96			class ActionAliasExecuteCommand(resource.ResourceCommand):
97			display_attributes = ['id', 'name', 'description']
98
99			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-09-14 05:21 UTC by Report Bug Copy Issue Report `resource` is re-defining a name which is already available in the outer-scope (previously defined on line `24`). 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...
100			super(ActionAliasExecuteCommand, self).__init__(
101			resource, 'execute',
102			'Execute the command text by finding a matching ActionAlias.',
103			args, *kwargs)
104
105			self.parser.add_argument('command_text',
106			metavar='command',
107			help=help)
108			self.parser.add_argument('--trace-tag', '--trace_tag',
109			help='A trace tag string to track execution later.',
110			dest='trace_tag', required=False)
111			self.parser.add_argument('--trace-id',
112			help='Existing trace id for this execution.',
113			dest='trace_id', required=False)
114			self.parser.add_argument('-a', '--async',
115			action='store_true', dest='async',
116			help='Do not wait for action to finish.')
117			self.parser.add_argument('-u', '--user', type=str, default=None,
118			help='User under which to run the action (admins only).')
119
120			@resource.add_auth_token_to_kwargs_from_cli
121			def run(self, args, **kwargs):
122			aliases = self.manager.get_all(**kwargs)
123			matches = match_command_to_alias(args.command_text, aliases)
124			if len(matches) > 1:
125			raise ActionAliasAmbiguityException("Too many matches for provided command",
126			matches=matches)
127			elif len(matches) == 0:
128			raise ActionAliasAmbiguityException("Could not locate an ActionAlias with a "
129			"matching format command", matches=matches)
130			match = matches[0]
131			action_alias_db = match[0]
132
133			if not action_alias_db.enabled:
134			raise ValueError('Action alias with name "%s" is disabled.' %
135			(action_alias_db.ref))
136
137			execution_parameters = extract_parameters_for_action_alias_db(
138			action_alias_db=action_alias_db,
139			format_str=matches[2],
140			param_stream=args.command_text)
141
142			execution = models.LiveAction()
143			execution.action = action_alias_db.action_ref
144			execution.parameters = execution_parameters
145			execution.user = args.user
146
147			if not args.trace_id and args.trace_tag:
148			execution.context = {'trace_context': {'trace_tag': args.trace_tag}}
149
150			if args.trace_id:
151			execution.context = {'trace_context': {'id_': args.trace_id}}
152
153			action_exec_mgr = self.app.client.managers['LiveAction']
154
155			execution = action_exec_mgr.create(execution, **kwargs)
156			execution = self._get_execution_result(execution=execution,
157			action_exec_mgr=action_exec_mgr,
158			args=args, **kwargs)
159			return execution
160
161			def run_and_print(self, args, **kwargs):
162			instances = self.run(args, **kwargs)
163			self.print_output(instances, table.MultiColumnTable,
164			attributes=args.attr, widths=args.width,
165			json=args.json, yaml=args.yaml)
166

StackStorm / st2

Pull Request — master (#2895)

ActionAliasExecuteCommand.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ActionAliasExecuteCommand.init() A