PackInstallCommand.run_and_print() - Code Metrics - Inspection of "Updating changelog" - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — plexxi-v2.2.1 ( 2317c4 )

unknown

created 2017-04-06 00:54 UTC

PackInstallCommand.run_and_print() B

↳ Parent: PackInstallCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
c	0
b	0
f	0
dl	0
loc	22
rs	8.9197

# 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

import editor
import yaml

from st2client.models import Config
from st2client.models import Pack
from st2client.models import LiveAction
from st2client.commands import resource
from st2client.commands.action import ActionRunCommandMixin
from st2client.formatters import table
from st2client.exceptions.operations import OperationFailureException
import st2client.utils.terminal as term
from st2client.utils import interactive


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': PackGetCommand
            })

        self.commands['show'] = PackShowCommand(self.resource, self.app, self.subparsers)
        self.commands['search'] = PackSearchCommand(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['register'] = PackRegisterCommand(self.resource, self.app, self.subparsers)
        self.commands['config'] = PackConfigCommand(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.parser.add_argument('-w', '--width', nargs='+', type=int, default=None,
                                       help='Set the width of columns in output.')

        detail_arg_grp = self.parser.add_mutually_exclusive_group()
        detail_arg_grp.add_argument('--attr', nargs='+',
                                    default=['name', 'description', 'version', 'author'],
                                    help=('List of attributes to include in the '
                                          'output. "all" or unspecified will '
                                          'return all attributes.'))
        detail_arg_grp.add_argument('-d', '--detail', action='store_true',
                                    help='Display full detail of the execution in table format.')

    @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

                # Suppress intermediate output in case output formatter is requested
                if args.json or args.yaml:
                    continue

                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:
            args.depth = 1
            self._print_execution_details(execution=execution, args=args, **kwargs)
            sys.exit(1)

        return self.app.client.managers['LiveAction'].get_by_id(parent_id)


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


class PackGetCommand(resource.ResourceGetCommand):
    pk_argument_name = 'ref'
    display_attributes = ['name', 'version', 'author', 'email', 'keywords', 'description']
    attribute_display_order = ['name', 'version', 'author', 'email', 'keywords', 'description']
    help_string = 'Get information about an installed pack.'


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
        help_string = ('Get information about an available %s from the index.' %
                       resource.get_display_name().lower())
        super(PackShowCommand, self).__init__(resource, 'show', help_string,
                                              *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 in Exchange, or a git repo URL.' %
                                 resource.get_plural_display_name().lower())
        self.parser.add_argument('--force',
                                 action='store_true',
                                 default=False,
                                 help='Force pack installation.')

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

    def run_and_print(self, args, **kwargs):
        instance = super(PackInstallCommand, self).run_and_print(args, **kwargs)

        # Hack to get a list of resolved references of installed packs
        packs = instance.result['tasks'][1]['result']['result']

        if len(packs) == 1:
            pack_instance = self.app.client.managers['Pack'].get_by_ref_or_id(packs[0])
            self.print_output(pack_instance, table.PropertyValueTable,
                              attributes=args.attr, json=args.json, yaml=args.yaml,
                              attribute_display_order=self.attribute_display_order)
        else:
            all_pack_instances = self.app.client.managers['Pack'].get_all()
            pack_instances = []

            for pack in all_pack_instances:
                if pack.name in packs:
                    pack_instances.append(pack)

            self.print_output(pack_instances, table.MultiColumnTable,
                              attributes=args.attr, widths=args.width,
                              json=args.json, yaml=args.yaml)


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)

    def run_and_print(self, args, **kwargs):
        all_pack_instances = self.app.client.managers['Pack'].get_all()

        super(PackRemoveCommand, self).run_and_print(args, **kwargs)

        packs = args.packs

        if len(packs) == 1:
            pack_instance = self.app.client.managers['Pack'].get_by_ref_or_id(packs[0])

            if pack_instance:
                raise OperationFailureException('Pack %s has not been removed properly', packs[0])

            removed_pack_instance = next((pack for pack in all_pack_instances
                                         if pack.name == packs[0]), None)

            self.print_output(removed_pack_instance, table.PropertyValueTable,
                              attributes=args.attr, json=args.json, yaml=args.yaml,
                              attribute_display_order=self.attribute_display_order)
        else:
            remaining_pack_instances = self.app.client.managers['Pack'].get_all()
            pack_instances = []

            for pack in all_pack_instances:
                if pack.name in packs:
                    pack_instances.append(pack)
                if pack in remaining_pack_instances:
                    raise OperationFailureException('Pack %s has not been removed properly',
                                                    pack.name)

            self.print_output(pack_instances, table.MultiColumnTable,
                              attributes=args.attr, widths=args.width,
                              json=args.json, yaml=args.yaml)


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='*',
                                 metavar='pack',
                                 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 the index for a %s with any attribute matching the query.'
            % 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)


class PackConfigCommand(resource.ResourceCommand):
    def __init__(self, resource, *args, **kwargs):
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(PackConfigCommand, self).__init__(resource, 'config',
              'Configure a %s based on config schema.' % resource.get_display_name().lower(),
              *args, **kwargs)

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

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        schema = self.app.client.managers['ConfigSchema'].get_by_ref_or_id(args.name, **kwargs)

        if not schema:
            msg = '%s "%s" doesn\'t exist or doesn\'t have config schema defined.'
            raise resource.ResourceNotFoundError(msg % (self.resource.get_display_name(),
                                                        args.name))

        config = interactive.InteractiveForm(schema.attributes).initiate_dialog()

        message = '---\nDo you want to preview the config in an editor before saving?'
        description = 'Secrets would be shown in plain text.'
        preview_dialog = interactive.Question(message, {'default': 'y', 'description': description})
        if preview_dialog.read() == 'y':
            try:
                contents = yaml.safe_dump(config, indent=4, default_flow_style=False)
                modified = editor.edit(contents=contents)
                config = yaml.safe_load(modified)
            except editor.EditorError as e:
                print(str(e))

        message = '---\nDo you want me to save it?'
        save_dialog = interactive.Question(message, {'default': 'y'})
        if save_dialog.read() == 'n':
            raise OperationFailureException('Interrupted')

        result = self.app.client.managers['Config'].update(Config(pack=args.name, values=config))

        return result

    def run_and_print(self, args, **kwargs):
        try:
            instance = self.run(args, **kwargs)
            if not instance:
                raise Exception("Configuration failed")
            self.print_output(instance, table.PropertyValueTable,
                              attributes=['all'], json=args.json, yaml=args.yaml)
        except (KeyboardInterrupt, SystemExit):
            raise OperationFailureException('Interrupted')
        except Exception as e:
            if self.app.client.debug:
                raise

            message = e.message or str(e)
            print('ERROR: %s' % (message))
            raise OperationFailureException(message)


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			import editor
19			import yaml
20
21			from st2client.models import Config
22			from st2client.models import Pack
23			from st2client.models import LiveAction
24			from st2client.commands import resource
25			from st2client.commands.action import ActionRunCommandMixin
26			from st2client.formatters import table
27			from st2client.exceptions.operations import OperationFailureException
28			import st2client.utils.terminal as term
29			from st2client.utils import interactive
30
31
32			LIVEACTION_STATUS_REQUESTED = 'requested'
33			LIVEACTION_STATUS_SCHEDULED = 'scheduled'
34			LIVEACTION_STATUS_DELAYED = 'delayed'
35			LIVEACTION_STATUS_RUNNING = 'running'
36			LIVEACTION_STATUS_SUCCEEDED = 'succeeded'
37			LIVEACTION_STATUS_FAILED = 'failed'
38			LIVEACTION_STATUS_TIMED_OUT = 'timeout'
39			LIVEACTION_STATUS_ABANDONED = 'abandoned'
40			LIVEACTION_STATUS_CANCELING = 'canceling'
41			LIVEACTION_STATUS_CANCELED = 'canceled'
42
43			LIVEACTION_COMPLETED_STATES = [
44			LIVEACTION_STATUS_SUCCEEDED,
45			LIVEACTION_STATUS_FAILED,
46			LIVEACTION_STATUS_TIMED_OUT,
47			LIVEACTION_STATUS_CANCELED,
48			LIVEACTION_STATUS_ABANDONED
49			]
50
51
52			class PackBranch(resource.ResourceBranch):
53			def __init__(self, description, app, subparsers, parent_parser=None):
54			super(PackBranch, self).__init__(
55			Pack, description, app, subparsers,
56			parent_parser=parent_parser,
57			read_only=True,
58			commands={
59			'list': PackListCommand,
60			'get': PackGetCommand
61			})
62
63			self.commands['show'] = PackShowCommand(self.resource, self.app, self.subparsers)
64			self.commands['search'] = PackSearchCommand(self.resource, self.app, self.subparsers)
65			self.commands['install'] = PackInstallCommand(self.resource, self.app, self.subparsers)
66			self.commands['remove'] = PackRemoveCommand(self.resource, self.app, self.subparsers)
67			self.commands['register'] = PackRegisterCommand(self.resource, self.app, self.subparsers)
68			self.commands['config'] = PackConfigCommand(self.resource, self.app, self.subparsers)
69
70
71			class PackResourceCommand(resource.ResourceCommand):
72			def run_and_print(self, args, **kwargs):
73			try:
74			instance = self.run(args, **kwargs)
75			if not instance:
76			raise resource.ResourceNotFoundError("No matching items found")
77			self.print_output(instance, table.PropertyValueTable,
78			attributes=['all'], json=args.json, yaml=args.yaml)
79			except resource.ResourceNotFoundError:
80			print("No matching items found")
81			except Exception as e:
82			message = e.message or str(e)
83			print('ERROR: %s' % (message))
84			raise OperationFailureException(message)
85
86
87			class PackAsyncCommand(ActionRunCommandMixin, resource.ResourceCommand):
88			def __init__(self, args, *kwargs):
89			super(PackAsyncCommand, self).__init__(args, *kwargs)
90
91			self.parser.add_argument('-w', '--width', nargs='+', type=int, default=None,
92			help='Set the width of columns in output.')
93
94			detail_arg_grp = self.parser.add_mutually_exclusive_group()
95			detail_arg_grp.add_argument('--attr', nargs='+',
96			default=['name', 'description', 'version', 'author'],
97			help=('List of attributes to include in the '
98			'output. "all" or unspecified will '
99			'return all attributes.'))
100			detail_arg_grp.add_argument('-d', '--detail', action='store_true',
101			help='Display full detail of the execution in table format.')
102
103			@resource.add_auth_token_to_kwargs_from_cli
104			def run_and_print(self, args, **kwargs):
105			instance = self.run(args, **kwargs)
106			if not instance:
107			raise Exception('Server did not create instance.')
108
109			parent_id = instance.execution_id
110
111			stream_mgr = self.app.client.managers['Stream']
112
113			execution = None
114
115			with term.TaskIndicator() as indicator:
116			events = ['st2.execution__create', 'st2.execution__update']
117			for event in stream_mgr.listen(events, **kwargs):
118			execution = LiveAction(**event)
119
120			if execution.id == parent_id \
121			and execution.status in LIVEACTION_COMPLETED_STATES:
122			break
123
124			# Suppress intermediate output in case output formatter is requested
125			if args.json or args.yaml:
126			continue
127
128			if getattr(execution, 'parent', None) == parent_id:
129			status = execution.status
130			name = execution.context['chain']['name']
131
132			if status == LIVEACTION_STATUS_SCHEDULED:
133			indicator.add_stage(status, name)
134			if status == LIVEACTION_STATUS_RUNNING:
135			indicator.update_stage(status, name)
136			if status in LIVEACTION_COMPLETED_STATES:
137			indicator.finish_stage(status, name)
138
139			if execution and execution.status == LIVEACTION_STATUS_FAILED:
140			args.depth = 1
141			self._print_execution_details(execution=execution, args=args, **kwargs)
142			sys.exit(1)
143
144			return self.app.client.managers['LiveAction'].get_by_id(parent_id)
145
146
147			class PackListCommand(resource.ResourceListCommand):
148			display_attributes = ['ref', 'name', 'description', 'version', 'author']
149			attribute_display_order = ['ref', 'name', 'description', 'version', 'author']
150
151
152			class PackGetCommand(resource.ResourceGetCommand):
153			pk_argument_name = 'ref'
154			display_attributes = ['name', 'version', 'author', 'email', 'keywords', 'description']
155			attribute_display_order = ['name', 'version', 'author', 'email', 'keywords', 'description']
156			help_string = 'Get information about an installed pack.'
157
158
159			class PackShowCommand(PackResourceCommand):
160			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-12-18 21:45 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 `24`). 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...
161			help_string = ('Get information about an available %s from the index.' %
162			resource.get_display_name().lower())
163			super(PackShowCommand, self).__init__(resource, 'show', help_string,
164			args, *kwargs)
165
166			self.parser.add_argument('pack',
167			help='Name of the %s to show.' %
168			resource.get_display_name().lower())
169
170			@resource.add_auth_token_to_kwargs_from_cli
171			def run(self, args, **kwargs):
172			return self.manager.search(args, **kwargs)
173
174
175			class PackInstallCommand(PackAsyncCommand):
176			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-12-18 21:45 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 `24`). 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...
177			super(PackInstallCommand, self).__init__(resource, 'install',
178			'Install new %s.' % resource.get_plural_display_name().lower(),
179			args, *kwargs)
180
181			self.parser.add_argument('packs',
182			nargs='+',
183			metavar='pack',
184			help='Name of the %s in Exchange, or a git repo URL.' %
185			resource.get_plural_display_name().lower())
186			self.parser.add_argument('--force',
187			action='store_true',
188			default=False,
189			help='Force pack installation.')
190
191			@resource.add_auth_token_to_kwargs_from_cli
192			def run(self, args, **kwargs):
193			return self.manager.install(args.packs, force=args.force, **kwargs)
194
195			def run_and_print(self, args, **kwargs):
196			instance = super(PackInstallCommand, self).run_and_print(args, **kwargs)
197
198			# Hack to get a list of resolved references of installed packs
199			packs = instance.result['tasks'][1]['result']['result']
200
201			if len(packs) == 1:
202			pack_instance = self.app.client.managers['Pack'].get_by_ref_or_id(packs[0])
203			self.print_output(pack_instance, table.PropertyValueTable,
204			attributes=args.attr, json=args.json, yaml=args.yaml,
205			attribute_display_order=self.attribute_display_order)
206			else:
207			all_pack_instances = self.app.client.managers['Pack'].get_all()
208			pack_instances = []
209
210			for pack in all_pack_instances:
211			if pack.name in packs:
212			pack_instances.append(pack)
213
214			self.print_output(pack_instances, table.MultiColumnTable,
215			attributes=args.attr, widths=args.width,
216			json=args.json, yaml=args.yaml)
217
218
219			class PackRemoveCommand(PackAsyncCommand):
220			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-11-04 10:06 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 `24`). 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...
221			super(PackRemoveCommand, self).__init__(resource, 'remove',
222			'Remove %s.' % resource.get_plural_display_name().lower(),
223			args, *kwargs)
224
225			self.parser.add_argument('packs',
226			nargs='+',
227			metavar='pack',
228			help='Name of the %s to remove.' %
229			resource.get_plural_display_name().lower())
230
231			@resource.add_auth_token_to_kwargs_from_cli
232			def run(self, args, **kwargs):
233			return self.manager.remove(args.packs, **kwargs)
234
235			def run_and_print(self, args, **kwargs):
236			all_pack_instances = self.app.client.managers['Pack'].get_all()
237
238			super(PackRemoveCommand, self).run_and_print(args, **kwargs)
239
240			packs = args.packs
241
242			if len(packs) == 1:
243			pack_instance = self.app.client.managers['Pack'].get_by_ref_or_id(packs[0])
244
245			if pack_instance:
246			raise OperationFailureException('Pack %s has not been removed properly', packs[0])
247
248			removed_pack_instance = next((pack for pack in all_pack_instances
249			if pack.name == packs[0]), None)
250
251			self.print_output(removed_pack_instance, table.PropertyValueTable,
252			attributes=args.attr, json=args.json, yaml=args.yaml,
253			attribute_display_order=self.attribute_display_order)
254			else:
255			remaining_pack_instances = self.app.client.managers['Pack'].get_all()
256			pack_instances = []
257
258			for pack in all_pack_instances:
259			if pack.name in packs:
260			pack_instances.append(pack)
261			if pack in remaining_pack_instances:
262			raise OperationFailureException('Pack %s has not been removed properly',
263			pack.name)
264
265			self.print_output(pack_instances, table.MultiColumnTable,
266			attributes=args.attr, widths=args.width,
267			json=args.json, yaml=args.yaml)
268
269
270			class PackRegisterCommand(PackResourceCommand):
271			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-12-18 21:45 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 `24`). 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...
272			super(PackRegisterCommand, self).__init__(resource, 'register',
273			'Register a %s: sync all file changes with DB.' % resource.get_display_name().lower(),
274			args, *kwargs)
275
276			self.parser.add_argument('packs',
277			nargs='*',
278			metavar='pack',
279			help='Name of the %s(s) to register.' %
280			resource.get_display_name().lower())
281
282			self.parser.add_argument('--types',
283			nargs='+',
284			help='Types of content to register.')
285
286			@resource.add_auth_token_to_kwargs_from_cli
287			def run(self, args, **kwargs):
288			return self.manager.register(args.packs, args.types, **kwargs)
289
290
291			class PackSearchCommand(resource.ResourceTableCommand):
292			display_attributes = ['name', 'description', 'version', 'author']
293			attribute_display_order = ['name', 'description', 'version', 'author']
294
295			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-12-18 21:45 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 `24`). 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...
296			super(PackSearchCommand, self).__init__(resource, 'search',
297			'Search the index for a %s with any attribute matching the query.'
298			% resource.get_display_name().lower(),
299			args, *kwargs
300			)
301
302			self.parser.add_argument('query',
303			help='Search query.')
304
305			@resource.add_auth_token_to_kwargs_from_cli
306			def run(self, args, **kwargs):
307			return self.manager.search(args, **kwargs)
308
309
310			class PackConfigCommand(resource.ResourceCommand):
311			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-12-18 21:45 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 `24`). 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...
312			super(PackConfigCommand, self).__init__(resource, 'config',
313			'Configure a %s based on config schema.' % resource.get_display_name().lower(),
314			args, *kwargs)
315
316			self.parser.add_argument('name',
317			help='Name of the %s(s) to configure.' %
318			resource.get_display_name().lower())
319
320			@resource.add_auth_token_to_kwargs_from_cli
321			def run(self, args, **kwargs):
322			schema = self.app.client.managers['ConfigSchema'].get_by_ref_or_id(args.name, **kwargs)
323
324			if not schema:
325			msg = '%s "%s" doesn\'t exist or doesn\'t have config schema defined.'
326			raise resource.ResourceNotFoundError(msg % (self.resource.get_display_name(),
327			args.name))
328
329			config = interactive.InteractiveForm(schema.attributes).initiate_dialog()
330
331			message = '---\nDo you want to preview the config in an editor before saving?'
332			description = 'Secrets would be shown in plain text.'
333			preview_dialog = interactive.Question(message, {'default': 'y', 'description': description})
334			if preview_dialog.read() == 'y':
335			try:
336			contents = yaml.safe_dump(config, indent=4, default_flow_style=False)
337			modified = editor.edit(contents=contents)
338			config = yaml.safe_load(modified)
339			except editor.EditorError as e:
340			print(str(e))
341
342			message = '---\nDo you want me to save it?'
343			save_dialog = interactive.Question(message, {'default': 'y'})
344			if save_dialog.read() == 'n':
345			raise OperationFailureException('Interrupted')
346
347			result = self.app.client.managers['Config'].update(Config(pack=args.name, values=config))
348
349			return result
350
351			def run_and_print(self, args, **kwargs):
352			try:
353			instance = self.run(args, **kwargs)
354			if not instance:
355			raise Exception("Configuration failed")
356			self.print_output(instance, table.PropertyValueTable,
357			attributes=['all'], json=args.json, yaml=args.yaml)
358			except (KeyboardInterrupt, SystemExit):
359			raise OperationFailureException('Interrupted')
360			except Exception as e:
361			if self.app.client.debug:
362			raise
363
364			message = e.message or str(e)
365			print('ERROR: %s' % (message))
366			raise OperationFailureException(message)
367

Plexxi / st2

GitHub Access Token became invalid

Push — plexxi-v2.2.1 ( 2317c4 )

PackInstallCommand.run_and_print() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like