PackSearchCommand.__init__() - Code Metrics - Inspection of "Pack management" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2842)

by Edward

created 2016-10-06 13:28 UTC

PackSearchCommand.init() A

↳ Parent: PackSearchCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	7
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 st2client.models import Pack
from st2client.commands import resource
from st2client.commands.action import ActionRunCommandMixin
from st2client.commands.noop import NoopCommand
from st2client.formatters import table
from st2client.exceptions.operations import OperationFailureException
import st2client.utils.terminal as term


LIVEACTION_STATUS_REQUESTED = 'requested'
LIVEACTION_STATUS_SCHEDULED = 'scheduled'
LIVEACTION_STATUS_DELAYED = 'delayed'
LIVEACTION_STATUS_RUNNING = 'running'
LIVEACTION_STATUS_SUCCEEDED = 'succeeded'
LIVEACTION_STATUS_FAILED = 'failed'
LIVEACTION_STATUS_TIMED_OUT = 'timeout'
LIVEACTION_STATUS_ABANDONED = 'abandoned'
LIVEACTION_STATUS_CANCELING = 'canceling'
LIVEACTION_STATUS_CANCELED = 'canceled'

LIVEACTION_COMPLETED_STATES = [
    LIVEACTION_STATUS_SUCCEEDED,
    LIVEACTION_STATUS_FAILED,
    LIVEACTION_STATUS_TIMED_OUT,
    LIVEACTION_STATUS_CANCELED,
    LIVEACTION_STATUS_ABANDONED
]


class PackBranch(resource.ResourceBranch):
    def __init__(self, description, app, subparsers, parent_parser=None):
        super(PackBranch, self).__init__(
            Pack, description, app, subparsers,
            parent_parser=parent_parser,
            read_only=True,
            commands={
                'list': PackListCommand,
                'get': NoopCommand
            })

        self.commands['register'] = PackRegisterCommand(self.resource, self.app, self.subparsers)
        self.commands['create'] = PackCreateCommand(self.resource, self.app, self.subparsers)
        self.commands['install'] = PackInstallCommand(self.resource, self.app, self.subparsers)
        self.commands['remove'] = PackRemoveCommand(self.resource, self.app, self.subparsers)
        self.commands['search'] = PackSearchCommand(self.resource, self.app, self.subparsers)
        self.commands['show'] = PackShowCommand(self.resource, self.app, self.subparsers)


class PackResourceCommand(resource.ResourceCommand):
    def run_and_print(self, args, **kwargs):
        try:
            instance = self.run(args, **kwargs)
            if not instance:
                raise Exception('Server did not create instance.')
            self.print_output(instance, table.PropertyValueTable,
                              attributes=['all'], json=args.json, yaml=args.yaml)
        except Exception as e:
            message = e.message or str(e)
            print('ERROR: %s' % (message))
            raise OperationFailureException(message)


class PackAsyncCommand(ActionRunCommandMixin, resource.ResourceCommand):
    def run_and_print(self, args, **kwargs):
        instance = self.run(args, **kwargs)
        if not instance:
            raise Exception('Server did not create instance.')

        parent_id = instance.execution_id

        stream_mgr = self.app.client.managers['Stream']

        with term.TaskIndicator() as indicator:
            for execution in stream_mgr.listen(['st2.execution__create', 'st2.execution__update']):
                if execution['id'] == parent_id \
                        and execution['status'] in LIVEACTION_COMPLETED_STATES:
                    break

                if execution.get('parent', None) == parent_id:
                    status = execution['status']
                    name = execution['context']['chain']['name']

                    if status == LIVEACTION_STATUS_SCHEDULED:
                        indicator.add_stage(status, name)
                    if status == LIVEACTION_STATUS_RUNNING:
                        indicator.update_stage(status, name)
                    if status in LIVEACTION_COMPLETED_STATES:
                        indicator.finish_stage(status, name)


class PackListCommand(resource.ResourceListCommand):
    display_attributes = ['name', 'description', 'version', 'author']
    attribute_display_order = ['name', 'description', 'version', 'author']


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

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(PackShowCommand, self).__init__(resource, 'show',
              'Get information about a %s from the index.' % resource.get_display_name().lower(),
              *args, **kwargs)

        self.parser.add_argument('pack',
                                 help='Name of the %s to show.' %
                                 resource.get_display_name().lower())

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        return self.manager.search(args, **kwargs)


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

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(PackInstallCommand, self).__init__(resource, 'install',
            'Install new %s.' % resource.get_plural_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('packs',
                                 nargs='+',
                                 metavar='pack',
                                 help='Name of the %s to install.' %
                                 resource.get_plural_display_name().lower())

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        return self.manager.install(args.packs, **kwargs)


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

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(PackRemoveCommand, self).__init__(resource, 'remove',
            'Remove %s.' % resource.get_plural_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('packs',
                                 nargs='+',
                                 metavar='pack',
                                 help='Name of the %s to remove.' %
                                 resource.get_plural_display_name().lower())

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        return self.manager.remove(args.packs, **kwargs)


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

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(PackCreateCommand, self).__init__(resource, 'create',
            'Create a template for a new %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('name',
                                 help='Name of the %s to create.' %
                                 resource.get_display_name().lower())
        self.parser.add_argument('--description',
                                 required=True,
                                 help='Description of the %s.' %
                                 resource.get_display_name().lower())
        self.parser.add_argument('--keywords',
                                 nargs='+',
                                 help='Keywords describing the %s.' %
                                 resource.get_display_name().lower())
        self.parser.add_argument('--version',
                                 help='Version of the %s.' %
                                 resource.get_display_name().lower())
        self.parser.add_argument('--author',
                                 required=True,
                                 help='Author of the %s.' %
                                 resource.get_display_name().lower())
        self.parser.add_argument('--email',
                                 help='%s author\'s email.' %
                                 resource.get_display_name())

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        return self.manager.create(args, **kwargs)


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

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(PackRegisterCommand, self).__init__(resource, 'register',
              'Register a %s: sync all file changes with DB.' % resource.get_display_name().lower(),
              *args, **kwargs)

        self.parser.add_argument('--types',
                                 nargs='+',
                                 help='Name of the %s to register.' %
                                 resource.get_display_name().lower())

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        return self.manager.register(args.types, **kwargs)


class PackSearchCommand(resource.ResourceTableCommand):
    display_attributes = ['name', 'description', 'version', 'author']
    attribute_display_order = ['name', 'description', 'version', 'author']

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

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(PackSearchCommand, self).__init__(resource, 'search',
            'Search for a %s in the directory.' % resource.get_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('query',
                                 help='Search query.')

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        return self.manager.search(args, **kwargs)


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 st2client.models import Pack
17			from st2client.commands import resource
18			from st2client.commands.action import ActionRunCommandMixin
19			from st2client.commands.noop import NoopCommand
20			from st2client.formatters import table
21			from st2client.exceptions.operations import OperationFailureException
22			import st2client.utils.terminal as term
23
24
25			LIVEACTION_STATUS_REQUESTED = 'requested'
26			LIVEACTION_STATUS_SCHEDULED = 'scheduled'
27			LIVEACTION_STATUS_DELAYED = 'delayed'
28			LIVEACTION_STATUS_RUNNING = 'running'
29			LIVEACTION_STATUS_SUCCEEDED = 'succeeded'
30			LIVEACTION_STATUS_FAILED = 'failed'
31			LIVEACTION_STATUS_TIMED_OUT = 'timeout'
32			LIVEACTION_STATUS_ABANDONED = 'abandoned'
33			LIVEACTION_STATUS_CANCELING = 'canceling'
34			LIVEACTION_STATUS_CANCELED = 'canceled'
35
36			LIVEACTION_COMPLETED_STATES = [
37			LIVEACTION_STATUS_SUCCEEDED,
38			LIVEACTION_STATUS_FAILED,
39			LIVEACTION_STATUS_TIMED_OUT,
40			LIVEACTION_STATUS_CANCELED,
41			LIVEACTION_STATUS_ABANDONED
42			]
43
44
45			class PackBranch(resource.ResourceBranch):
46			def __init__(self, description, app, subparsers, parent_parser=None):
47			super(PackBranch, self).__init__(
48			Pack, description, app, subparsers,
49			parent_parser=parent_parser,
50			read_only=True,
51			commands={
52			'list': PackListCommand,
53			'get': NoopCommand
54			})
55
56			self.commands['register'] = PackRegisterCommand(self.resource, self.app, self.subparsers)
57			self.commands['create'] = PackCreateCommand(self.resource, self.app, self.subparsers)
58			self.commands['install'] = PackInstallCommand(self.resource, self.app, self.subparsers)
59			self.commands['remove'] = PackRemoveCommand(self.resource, self.app, self.subparsers)
60			self.commands['search'] = PackSearchCommand(self.resource, self.app, self.subparsers)
61			self.commands['show'] = PackShowCommand(self.resource, self.app, self.subparsers)
62
63
64			class PackResourceCommand(resource.ResourceCommand):
65			def run_and_print(self, args, **kwargs):
66			try:
67			instance = self.run(args, **kwargs)
68			if not instance:
69			raise Exception('Server did not create instance.')
70			self.print_output(instance, table.PropertyValueTable,
71			attributes=['all'], json=args.json, yaml=args.yaml)
72			except Exception as e:
73			message = e.message or str(e)
74			print('ERROR: %s' % (message))
75			raise OperationFailureException(message)
76
77
78			class PackAsyncCommand(ActionRunCommandMixin, resource.ResourceCommand):
79			def run_and_print(self, args, **kwargs):
80			instance = self.run(args, **kwargs)
81			if not instance:
82			raise Exception('Server did not create instance.')
83
84			parent_id = instance.execution_id
85
86			stream_mgr = self.app.client.managers['Stream']
87
88			with term.TaskIndicator() as indicator:
89			for execution in stream_mgr.listen(['st2.execution__create', 'st2.execution__update']):
90			if execution['id'] == parent_id \
91			and execution['status'] in LIVEACTION_COMPLETED_STATES:
92			break
93
94			if execution.get('parent', None) == parent_id:
95			status = execution['status']
96			name = execution['context']['chain']['name']
97
98			if status == LIVEACTION_STATUS_SCHEDULED:
99			indicator.add_stage(status, name)
100			if status == LIVEACTION_STATUS_RUNNING:
101			indicator.update_stage(status, name)
102			if status in LIVEACTION_COMPLETED_STATES:
103			indicator.finish_stage(status, name)
104
105
106			class PackListCommand(resource.ResourceListCommand):
107			display_attributes = ['name', 'description', 'version', 'author']
108			attribute_display_order = ['name', 'description', 'version', 'author']
109
110
111			class PackShowCommand(PackResourceCommand):
112			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-08-02 05:26 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 `17`). 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...
113			super(PackShowCommand, self).__init__(resource, 'show',
114			'Get information about a %s from the index.' % resource.get_display_name().lower(),
115			args, *kwargs)
116
117			self.parser.add_argument('pack',
118			help='Name of the %s to show.' %
119			resource.get_display_name().lower())
120
121			@resource.add_auth_token_to_kwargs_from_cli
122			def run(self, args, **kwargs):
123			return self.manager.search(args, **kwargs)
124
125
126			class PackInstallCommand(PackAsyncCommand):
127			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-10-03 07:23 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 `17`). 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...
128			super(PackInstallCommand, self).__init__(resource, 'install',
129			'Install new %s.' % resource.get_plural_display_name().lower(),
130			args, *kwargs)
131
132			self.parser.add_argument('packs',
133			nargs='+',
134			metavar='pack',
135			help='Name of the %s to install.' %
136			resource.get_plural_display_name().lower())
137
138			@resource.add_auth_token_to_kwargs_from_cli
139			def run(self, args, **kwargs):
140			return self.manager.install(args.packs, **kwargs)
141
142
143			class PackRemoveCommand(PackAsyncCommand):
144			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-10-03 07:23 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 `17`). 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...
145			super(PackRemoveCommand, self).__init__(resource, 'remove',
146			'Remove %s.' % resource.get_plural_display_name().lower(),
147			args, *kwargs)
148
149			self.parser.add_argument('packs',
150			nargs='+',
151			metavar='pack',
152			help='Name of the %s to remove.' %
153			resource.get_plural_display_name().lower())
154
155			@resource.add_auth_token_to_kwargs_from_cli
156			def run(self, args, **kwargs):
157			return self.manager.remove(args.packs, **kwargs)
158
159
160			class PackCreateCommand(PackAsyncCommand):
161			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-08-02 05:26 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 `17`). 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...
162			super(PackCreateCommand, self).__init__(resource, 'create',
163			'Create a template for a new %s.' % resource.get_display_name().lower(),
164			args, *kwargs)
165
166			self.parser.add_argument('name',
167			help='Name of the %s to create.' %
168			resource.get_display_name().lower())
169			self.parser.add_argument('--description',
170			required=True,
171			help='Description of the %s.' %
172			resource.get_display_name().lower())
173			self.parser.add_argument('--keywords',
174			nargs='+',
175			help='Keywords describing the %s.' %
176			resource.get_display_name().lower())
177			self.parser.add_argument('--version',
178			help='Version of the %s.' %
179			resource.get_display_name().lower())
180			self.parser.add_argument('--author',
181			required=True,
182			help='Author of the %s.' %
183			resource.get_display_name().lower())
184			self.parser.add_argument('--email',
185			help='%s author\'s email.' %
186			resource.get_display_name())
187
188			@resource.add_auth_token_to_kwargs_from_cli
189			def run(self, args, **kwargs):
190			return self.manager.create(args, **kwargs)
191
192
193			class PackRegisterCommand(PackResourceCommand):
194			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-10-03 07:23 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 `17`). 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...
195			super(PackRegisterCommand, self).__init__(resource, 'register',
196			'Register a %s: sync all file changes with DB.' % resource.get_display_name().lower(),
197			args, *kwargs)
198
199			self.parser.add_argument('--types',
200			nargs='+',
201			help='Name of the %s to register.' %
202			resource.get_display_name().lower())
203
204			@resource.add_auth_token_to_kwargs_from_cli
205			def run(self, args, **kwargs):
206			return self.manager.register(args.types, **kwargs)
207
208
209			class PackSearchCommand(resource.ResourceTableCommand):
210			display_attributes = ['name', 'description', 'version', 'author']
211			attribute_display_order = ['name', 'description', 'version', 'author']
212
213			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-10-06 09:19 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 `17`). 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...
214			super(PackSearchCommand, self).__init__(resource, 'search',
215			'Search for a %s in the directory.' % resource.get_display_name().lower(),
216			args, *kwargs)
217
218			self.parser.add_argument('query',
219			help='Search query.')
220
221			@resource.add_auth_token_to_kwargs_from_cli
222			def run(self, args, **kwargs):
223			return self.manager.search(args, **kwargs)
224

StackStorm / st2

Pull Request — master (#2842)

PackSearchCommand.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

PackSearchCommand.init() A