ActionAliasExecuteCommand.run() - 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:52 UTC

ActionAliasExecuteCommand.run() C

↳ Parent: ActionAliasExecuteCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	1

Metric	Value
cc	7
c	3
b	0
f	1
dl	0
loc	39
rs	5.5

# 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
from st2common.util.actionalias_matching import match_command_to_alias

from st2client import models
from st2client.models.action_alias import ActionAlias
from st2client.commands.action import ActionRunCommandMixin
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(
            self.resource, 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(ActionRunCommandMixin,
                                resource.ResourceCommand):
    display_attributes = ['id', 'name', 'description']
    poll_interval = 2
    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('-h', '--help',
                                 action='store_true', dest='help',
                                 help='Print usage for the given action.')
        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(
            format_str=match[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


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

StackStorm / st2

Pull Request — master (#2895)

ActionAliasExecuteCommand.run() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like