PackResourceCommand.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-20 23:05 UTC

PackResourceCommand.run_and_print() A

↳ Parent: PackResourceCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	4
c	1
b	0
f	0
dl	0
loc	13
rs	9.2

# 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['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 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['install'] = PackInstallCommand(self.resource, self.app, self.subparsers)
61			self.commands['remove'] = PackRemoveCommand(self.resource, self.app, self.subparsers)
62			self.commands['search'] = PackSearchCommand(self.resource, self.app, self.subparsers)
63			self.commands['show'] = PackShowCommand(self.resource, self.app, self.subparsers)
64
65
66			class PackResourceCommand(resource.ResourceCommand):
67			def run_and_print(self, args, **kwargs):
68			try:
69			instance = self.run(args, **kwargs)
70			if not instance:
71			raise resource.ResourceNotFoundError("No matching items found")
72			self.print_output(instance, table.PropertyValueTable,
73			attributes=['all'], json=args.json, yaml=args.yaml)
74			except resource.ResourceNotFoundError:
75			print("No matching items found")
76			except Exception as e:
77			message = e.message or str(e)
78			print('ERROR: %s' % (message))
79			raise OperationFailureException(message)
80
81
82			class PackAsyncCommand(ActionRunCommandMixin, resource.ResourceCommand):
83			def __init__(self, args, *kwargs):
84			super(PackAsyncCommand, self).__init__(args, *kwargs)
85
86			self._add_common_options()
87
88			@resource.add_auth_token_to_kwargs_from_cli
89			def run_and_print(self, args, **kwargs):
90			instance = self.run(args, **kwargs)
91			if not instance:
92			raise Exception('Server did not create instance.')
93
94			parent_id = instance.execution_id
95
96			stream_mgr = self.app.client.managers['Stream']
97
98			execution = None
99
100			with term.TaskIndicator() as indicator:
101			events = ['st2.execution__create', 'st2.execution__update']
102			for event in stream_mgr.listen(events, **kwargs):
103			execution = LiveAction(**event)
104
105			if execution.id == parent_id \
106			and execution.status in LIVEACTION_COMPLETED_STATES:
107			break
108
109			if getattr(execution, 'parent', None) == parent_id:
110			status = execution.status
111			name = execution.context['chain']['name']
112
113			if status == LIVEACTION_STATUS_SCHEDULED:
114			indicator.add_stage(status, name)
115			if status == LIVEACTION_STATUS_RUNNING:
116			indicator.update_stage(status, name)
117			if status in LIVEACTION_COMPLETED_STATES:
118			indicator.finish_stage(status, name)
119
120			if execution and execution.status == LIVEACTION_STATUS_FAILED:
121			self._print_execution_details(execution=execution, args=args, **kwargs)
122			sys.exit(1)
123
124
125			class PackListCommand(resource.ResourceListCommand):
126			display_attributes = ['name', 'description', 'version', 'author']
127			attribute_display_order = ['name', 'description', 'version', 'author']
128
129
130			class PackShowCommand(PackResourceCommand):
131			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...
132			super(PackShowCommand, self).__init__(resource, 'show',
133			'Get information about a %s from the index.' % resource.get_display_name().lower(),
134			args, *kwargs)
135
136			self.parser.add_argument('pack',
137			help='Name of the %s to show.' %
138			resource.get_display_name().lower())
139
140			@resource.add_auth_token_to_kwargs_from_cli
141			def run(self, args, **kwargs):
142			return self.manager.search(args, **kwargs)
143
144
145			class PackInstallCommand(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 `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...
147			super(PackInstallCommand, self).__init__(resource, 'install',
148			'Install new %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 install.' %
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.install(args.packs, **kwargs)
160
161
162			class PackRemoveCommand(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 `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...
164			super(PackRemoveCommand, self).__init__(resource, 'remove',
165			'Remove %s.' % resource.get_plural_display_name().lower(),
166			args, *kwargs)
167
168			self.parser.add_argument('packs',
169			nargs='+',
170			metavar='pack',
171			help='Name of the %s to remove.' %
172			resource.get_plural_display_name().lower())
173
174			@resource.add_auth_token_to_kwargs_from_cli
175			def run(self, args, **kwargs):
176			return self.manager.remove(args.packs, **kwargs)
177
178
179			class PackRegisterCommand(PackResourceCommand):
180			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...
181			super(PackRegisterCommand, self).__init__(resource, 'register',
182			'Register a %s: sync all file changes with DB.' % resource.get_display_name().lower(),
183			args, *kwargs)
184
185			self.parser.add_argument('--packs',
186			nargs='+',
187			help='Name of the %s(s) to register.' %
188			resource.get_display_name().lower())
189
190			self.parser.add_argument('--types',
191			nargs='+',
192			help='Types of content to register.')
193
194			@resource.add_auth_token_to_kwargs_from_cli
195			def run(self, args, **kwargs):
196			return self.manager.register(args.packs, args.types, **kwargs)
197
198
199			class PackSearchCommand(resource.ResourceTableCommand):
200			display_attributes = ['name', 'description', 'version', 'author']
201			attribute_display_order = ['name', 'description', 'version', 'author']
202
203			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...
204			super(PackSearchCommand, self).__init__(resource, 'search',
205			'Search for a %s in the directory.' % resource.get_display_name().lower(),
206			args, *kwargs)
207
208			self.parser.add_argument('query',
209			help='Search query.')
210
211			@resource.add_auth_token_to_kwargs_from_cli
212			def run(self, args, **kwargs):
213			return self.manager.search(args, **kwargs)
214

StackStorm / st2

Pull Request — master (#2842)

PackResourceCommand.run_and_print() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like