PackAsyncCommand.run_and_print() - 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-07 12:30 UTC

PackAsyncCommand.run_and_print() F

↳ Parent: PackAsyncCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	10
c	0
b	0
f	0
dl	0
loc	25
rs	3.1304

How to fix Complexity

# 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 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 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 resource.ResourceNotFoundError("No matching items found")
70			self.print_output(instance, table.PropertyValueTable,
71			attributes=['all'], json=args.json, yaml=args.yaml)
72			except resource.ResourceNotFoundError:
73			print("No matching items found")
74			except Exception as e:
75			message = e.message or str(e)
76			print('ERROR: %s' % (message))
77			raise OperationFailureException(message)
78
79
80			class PackAsyncCommand(ActionRunCommandMixin, resource.ResourceCommand):
81			def run_and_print(self, args, **kwargs):
82			instance = self.run(args, **kwargs)
83			if not instance:
84			raise Exception('Server did not create instance.')
85
86			parent_id = instance.execution_id
87
88			stream_mgr = self.app.client.managers['Stream']
89
90			with term.TaskIndicator() as indicator:
91			for execution in stream_mgr.listen(['st2.execution__create', 'st2.execution__update']):
92			if execution['id'] == parent_id \
93			and execution['status'] in LIVEACTION_COMPLETED_STATES:
94			break
95
96			if execution.get('parent', None) == parent_id:
97			status = execution['status']
98			name = execution['context']['chain']['name']
99
100			if status == LIVEACTION_STATUS_SCHEDULED:
101			indicator.add_stage(status, name)
102			if status == LIVEACTION_STATUS_RUNNING:
103			indicator.update_stage(status, name)
104			if status in LIVEACTION_COMPLETED_STATES:
105			indicator.finish_stage(status, name)
106
107
108			class PackListCommand(resource.ResourceListCommand):
109			display_attributes = ['name', 'description', 'version', 'author']
110			attribute_display_order = ['name', 'description', 'version', 'author']
111
112
113			class PackShowCommand(PackResourceCommand):
114			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-10-07 12:35 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...
115			super(PackShowCommand, self).__init__(resource, 'show',
116			'Get information about a %s from the index.' % resource.get_display_name().lower(),
117			args, *kwargs)
118
119			self.parser.add_argument('pack',
120			help='Name of the %s to show.' %
121			resource.get_display_name().lower())
122
123			@resource.add_auth_token_to_kwargs_from_cli
124			def run(self, args, **kwargs):
125			return self.manager.search(args, **kwargs)
126
127
128			class PackInstallCommand(PackAsyncCommand):
129			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...
130			super(PackInstallCommand, self).__init__(resource, 'install',
131			'Install new %s.' % resource.get_plural_display_name().lower(),
132			args, *kwargs)
133
134			self.parser.add_argument('packs',
135			nargs='+',
136			metavar='pack',
137			help='Name of the %s to install.' %
138			resource.get_plural_display_name().lower())
139
140			@resource.add_auth_token_to_kwargs_from_cli
141			def run(self, args, **kwargs):
142			return self.manager.install(args.packs, **kwargs)
143
144
145			class PackRemoveCommand(PackAsyncCommand):
146			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...
147			super(PackRemoveCommand, self).__init__(resource, 'remove',
148			'Remove %s.' % resource.get_plural_display_name().lower(),
149			args, *kwargs)
150
151			self.parser.add_argument('packs',
152			nargs='+',
153			metavar='pack',
154			help='Name of the %s to remove.' %
155			resource.get_plural_display_name().lower())
156
157			@resource.add_auth_token_to_kwargs_from_cli
158			def run(self, args, **kwargs):
159			return self.manager.remove(args.packs, **kwargs)
160
161
162			class PackCreateCommand(PackAsyncCommand):
163			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...
164			super(PackCreateCommand, self).__init__(resource, 'create',
165			'Create a template for a new %s.' % resource.get_display_name().lower(),
166			args, *kwargs)
167
168			self.parser.add_argument('name',
169			help='Name of the %s to create.' %
170			resource.get_display_name().lower())
171			self.parser.add_argument('--description',
172			required=True,
173			help='Description of the %s.' %
174			resource.get_display_name().lower())
175			self.parser.add_argument('--keywords',
176			nargs='+',
177			help='Keywords describing the %s.' %
178			resource.get_display_name().lower())
179			self.parser.add_argument('--version',
180			help='Version of the %s.' %
181			resource.get_display_name().lower())
182			self.parser.add_argument('--author',
183			required=True,
184			help='Author of the %s.' %
185			resource.get_display_name().lower())
186			self.parser.add_argument('--email',
187			help='%s author\'s email.' %
188			resource.get_display_name())
189
190			@resource.add_auth_token_to_kwargs_from_cli
191			def run(self, args, **kwargs):
192			return self.manager.create(args, **kwargs)
193
194
195			class PackRegisterCommand(PackResourceCommand):
196			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...
197			super(PackRegisterCommand, self).__init__(resource, 'register',
198			'Register a %s: sync all file changes with DB.' % resource.get_display_name().lower(),
199			args, *kwargs)
200
201			self.parser.add_argument('--types',
202			nargs='+',
203			help='Name of the %s to register.' %
204			resource.get_display_name().lower())
205
206			@resource.add_auth_token_to_kwargs_from_cli
207			def run(self, args, **kwargs):
208			return self.manager.register(args.types, **kwargs)
209
210
211			class PackSearchCommand(resource.ResourceTableCommand):
212			display_attributes = ['name', 'description', 'version', 'author']
213			attribute_display_order = ['name', 'description', 'version', 'author']
214
215			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...
216			super(PackSearchCommand, self).__init__(resource, 'search',
217			'Search for a %s in the directory.' % resource.get_display_name().lower(),
218			args, *kwargs)
219
220			self.parser.add_argument('query',
221			help='Search query.')
222
223			@resource.add_auth_token_to_kwargs_from_cli
224			def run(self, args, **kwargs):
225			return self.manager.search(args, **kwargs)
226

StackStorm / st2

Pull Request — master (#2842)

PackAsyncCommand.run_and_print() F

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like