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-19 22:35 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.

import sys

from st2client.models import Pack
from st2client.models import LiveAction
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 resource.ResourceNotFoundError("No matching items found")
            self.print_output(instance, table.PropertyValueTable,
                              attributes=['all'], json=args.json, yaml=args.yaml)
        except resource.ResourceNotFoundError:
            print("No matching items found")
        except Exception as e:
            message = e.message or str(e)
            print('ERROR: %s' % (message))
            raise OperationFailureException(message)


class PackAsyncCommand(ActionRunCommandMixin, resource.ResourceCommand):
    def __init__(self, *args, **kwargs):
        super(PackAsyncCommand, self).__init__(*args, **kwargs)

        self._add_common_options()

    @resource.add_auth_token_to_kwargs_from_cli
    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']

        execution = None

        with term.TaskIndicator() as indicator:
            events = ['st2.execution__create', 'st2.execution__update']
            for event in stream_mgr.listen(events, **kwargs):
                execution = LiveAction(**event)

                if execution.id == parent_id \
                        and execution.status in LIVEACTION_COMPLETED_STATES:
                    break

                if getattr(execution, '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)

        if execution and execution.status == LIVEACTION_STATUS_FAILED:
            self._print_execution_details(execution=execution, args=args, **kwargs)
            sys.exit(1)


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('--packs',
                                 nargs='+',
                                 help='Name of the %s(s) to register.' %
                                 resource.get_display_name().lower())

        self.parser.add_argument('--types',
                                 nargs='+',
                                 help='Types of content to register.')

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

StackStorm / st2

Pull Request — master (#2842)

PackSearchCommand.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

PackSearchCommand.init() A