ResourceBranch.__init__() - Code Metrics - Inspection of "Merge pull request #2674 from StackStorm/hook_up_r..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 58fbab...4c05e2 )

by Plexxi

created 2016-05-11 06:42 UTC

ResourceBranch.init() F

↳ Parent: ResourceBranch

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	10
dl	0
loc	46
rs	3.2727

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.

import os
import abc
import six
import json
import logging
import httplib
from functools import wraps

import yaml

from st2client import commands
from st2client.exceptions.operations import OperationFailureException
from st2client.formatters import table

ALLOWED_EXTS = ['.json', '.yaml', '.yml']
PARSER_FUNCS = {'.json': json.load, '.yml': yaml.safe_load, '.yaml': yaml.safe_load}
LOG = logging.getLogger(__name__)


def add_auth_token_to_kwargs_from_cli(func):
    @wraps(func)
    def decorate(*args, **kwargs):
        ns = args[1]
        if getattr(ns, 'token', None):
            kwargs['token'] = ns.token
        return func(*args, **kwargs)
    return decorate


class ResourceCommandError(Exception):
    pass


class ResourceNotFoundError(Exception):
    pass


class ResourceBranch(commands.Branch):

    def __init__(self, resource, description, app, subparsers,
                 parent_parser=None, read_only=False, commands=None,
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
                 has_disable=False):

        self.resource = resource
        super(ResourceBranch, self).__init__(
            self.resource.get_alias().lower(), description,
            app, subparsers, parent_parser=parent_parser)

        # Registers subcommands for managing the resource type.
        self.subparsers = self.parser.add_subparsers(
            help=('List of commands for managing %s.' %
                  self.resource.get_plural_display_name().lower()))

        # Resolves if commands need to be overridden.
        commands = commands or {}
        if 'list' not in commands:
            commands['list'] = ResourceListCommand
        if 'get' not in commands:
            commands['get'] = ResourceGetCommand
        if 'create' not in commands:
            commands['create'] = ResourceCreateCommand
        if 'update' not in commands:
            commands['update'] = ResourceUpdateCommand
        if 'delete' not in commands:
            commands['delete'] = ResourceDeleteCommand

        if 'enable' not in commands:
            commands['enable'] = ResourceEnableCommand

        if 'disable' not in commands:
            commands['disable'] = ResourceDisableCommand

        # Instantiate commands.
        args = [self.resource, self.app, self.subparsers]
        self.commands['list'] = commands['list'](*args)
        self.commands['get'] = commands['get'](*args)

        if not read_only:
            self.commands['create'] = commands['create'](*args)
            self.commands['update'] = commands['update'](*args)
            self.commands['delete'] = commands['delete'](*args)

        if has_disable:
            self.commands['enable'] = commands['enable'](*args)
            self.commands['disable'] = commands['disable'](*args)


@six.add_metaclass(abc.ABCMeta)
class ResourceCommand(commands.Command):
    pk_argument_name = None

    def __init__(self, resource, *args, **kwargs):

        has_token_opt = kwargs.pop('has_token_opt', True)

        super(ResourceCommand, self).__init__(*args, **kwargs)

        self.resource = resource

        if has_token_opt:
            self.parser.add_argument('-t', '--token', dest='token',
                                     help='Access token for user authentication. '
                                          'Get ST2_AUTH_TOKEN from the environment '
                                          'variables by default.')

        self.parser.add_argument('-j', '--json',
                                 action='store_true', dest='json',
                                 help='Prints output in JSON format.')

    @property
    def manager(self):
        return self.app.client.managers[self.resource.__name__]

    @property
    def arg_name_for_resource_id(self):
        resource_name = self.resource.get_display_name().lower()
        return '%s-id' % resource_name.replace(' ', '-')

    def print_not_found(self, name):
        print ('%s "%s" is not found.\n' %

               (self.resource.get_display_name(), name))

    def get_resource(self, name_or_id, **kwargs):
        pk_argument_name = self.pk_argument_name

        if pk_argument_name == 'name_or_id':
            instance = self.get_resource_by_name_or_id(name_or_id=name_or_id, **kwargs)
        elif pk_argument_name == 'ref_or_id':
            instance = self.get_resource_by_ref_or_id(ref_or_id=name_or_id, **kwargs)
        else:
            instance = self.get_resource_by_pk(pk=name_or_id, **kwargs)

        return instance

    def get_resource_by_pk(self, pk, **kwargs):
        """
        Retrieve resource by a primary key.
        """
        try:
            instance = self.manager.get_by_id(pk, **kwargs)
        except Exception as e:
            # Hack for "Unauthorized" exceptions, we do want to propagate those
            response = getattr(e, 'response', None)
            status_code = getattr(response, 'status_code', None)
            if status_code and status_code == httplib.UNAUTHORIZED:
                raise e

            instance = None

        return instance

    def get_resource_by_id(self, id, **kwargs):

        instance = self.get_resource_by_pk(pk=id, **kwargs)

        if not instance:
            message = ('Resource with id "%s" doesn\'t exist.' % (id))
            raise ResourceNotFoundError(message)
        return instance

    def get_resource_by_name(self, name, **kwargs):
        """
        Retrieve resource by name.
        """
        instance = self.manager.get_by_name(name, **kwargs)
        return instance

    def get_resource_by_name_or_id(self, name_or_id, **kwargs):
        instance = self.get_resource_by_name(name=name_or_id, **kwargs)
        if not instance:
            instance = self.get_resource_by_pk(pk=name_or_id, **kwargs)

        if not instance:
            message = ('Resource with id or name "%s" doesn\'t exist.' %
                       (name_or_id))
            raise ResourceNotFoundError(message)
        return instance

    def get_resource_by_ref_or_id(self, ref_or_id, **kwargs):
        instance = self.manager.get_by_ref_or_id(ref_or_id=ref_or_id, **kwargs)

        if not instance:
            message = ('Resource with id or reference "%s" doesn\'t exist.' %
                       (ref_or_id))
            raise ResourceNotFoundError(message)
        return instance

    @abc.abstractmethod
    def run(self, args, **kwargs):
        raise NotImplementedError

    @abc.abstractmethod
    def run_and_print(self, args, **kwargs):
        raise NotImplementedError

    def _get_metavar_for_argument(self, argument):
        return argument.replace('_', '-')

    def _get_help_for_argument(self, resource, argument):
        argument_display_name = argument.title()
        resource_display_name = resource.get_display_name().lower()

        if 'ref' in argument:
            result = ('Reference or ID of the %s.' % (resource_display_name))
        elif 'name_or_id' in argument:
            result = ('Name or ID of the %s.' % (resource_display_name))
        else:
            result = ('%s of the %s.' % (argument_display_name, resource_display_name))

        return result


class ResourceListCommand(ResourceCommand):
    display_attributes = ['id', 'name', 'description']

    def __init__(self, resource, *args, **kwargs):
        super(ResourceListCommand, self).__init__(resource, 'list',
            'Get the list of %s.' % resource.get_plural_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('-a', '--attr', nargs='+',
                                 default=self.display_attributes,
                                 help=('List of attributes to include in the '
                                       'output. "all" will return all '
                                       'attributes.'))
        self.parser.add_argument('-w', '--width', nargs='+', type=int,
                                 default=None,
                                 help=('Set the width of columns in output.'))

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

    def run_and_print(self, args, **kwargs):
        instances = self.run(args, **kwargs)
        self.print_output(instances, table.MultiColumnTable,
                          attributes=args.attr, widths=args.width,
                          json=args.json)


class ContentPackResourceListCommand(ResourceListCommand):
    """
    Base command class for use with resources which belong to a content pack.
    """
    def __init__(self, resource, *args, **kwargs):
        super(ContentPackResourceListCommand, self).__init__(resource,
                                                             *args, **kwargs)

        self.parser.add_argument('-p', '--pack', type=str,
                                 help=('Only return resources belonging to the'
                                       ' provided pack'))

    @add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        filters = {'pack': args.pack}
        filters.update(**kwargs)
        return self.manager.get_all(**filters)


class ResourceGetCommand(ResourceCommand):
    display_attributes = ['all']
    attribute_display_order = ['id', 'name', 'description']

    pk_argument_name = 'name_or_id'  # name of the attribute which stores resource PK

    def __init__(self, resource, *args, **kwargs):
        super(ResourceGetCommand, self).__init__(resource, 'get',
            'Get individual %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        argument = self.pk_argument_name
        metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
        help = self._get_help_for_argument(resource=resource,

                                           argument=self.pk_argument_name)

        self.parser.add_argument(argument,
                                 metavar=metavar,
                                 help=help)
        self.parser.add_argument('-a', '--attr', nargs='+',
                                 default=self.display_attributes,
                                 help=('List of attributes to include in the '
                                       'output. "all" or unspecified will '
                                       'return all attributes.'))

    @add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        resource_id = getattr(args, self.pk_argument_name, None)
        return self.get_resource_by_id(resource_id, **kwargs)

    def run_and_print(self, args, **kwargs):
        try:
            instance = self.run(args, **kwargs)
            self.print_output(instance, table.PropertyValueTable,
                              attributes=args.attr, json=args.json,
                              attribute_display_order=self.attribute_display_order)
        except ResourceNotFoundError:
            resource_id = getattr(args, self.pk_argument_name, None)
            self.print_not_found(resource_id)
            raise OperationFailureException('Resource %s not found.' % resource_id)


class ContentPackResourceGetCommand(ResourceGetCommand):
    """
    Command for retrieving a single resource which belongs to a content pack.

    Note: All the resources which belong to the content pack can either be
    retrieved by a reference or by an id.
    """

    attribute_display_order = ['id', 'pack', 'name', 'description']

    pk_argument_name = 'ref_or_id'

    def get_resource(self, ref_or_id, **kwargs):
        return self.get_resource_by_ref_or_id(ref_or_id=ref_or_id, **kwargs)


class ResourceCreateCommand(ResourceCommand):

    def __init__(self, resource, *args, **kwargs):
        super(ResourceCreateCommand, self).__init__(resource, 'create',
            'Create a new %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('file',
                                 help=('JSON/YAML file containing the %s to create.'
                                       % resource.get_display_name().lower()))

    @add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        data = _load_meta_file(args.file)
        instance = self.resource.deserialize(data)
        return self.manager.create(instance, **kwargs)

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


class ResourceUpdateCommand(ResourceCommand):
    pk_argument_name = 'name_or_id'

    def __init__(self, resource, *args, **kwargs):
        super(ResourceUpdateCommand, self).__init__(resource, 'update',
            'Updating an existing %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        argument = self.pk_argument_name
        metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
        help = self._get_help_for_argument(resource=resource,

                                           argument=self.pk_argument_name)

        self.parser.add_argument(argument,
                                 metavar=metavar,
                                 help=help)
        self.parser.add_argument('file',
                                 help=('JSON/YAML file containing the %s to update.'
                                       % resource.get_display_name().lower()))

    @add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        resource_id = getattr(args, self.pk_argument_name, None)
        instance = self.get_resource(resource_id, **kwargs)
        data = _load_meta_file(args.file)
        modified_instance = self.resource.deserialize(data)

        if not getattr(modified_instance, 'id', None):
            modified_instance.id = instance.id
        else:
            if modified_instance.id != instance.id:
                raise Exception('The value for the %s id in the JSON/YAML file '
                                'does not match the ID provided in the '
                                'command line arguments.' %
                                self.resource.get_display_name().lower())
        return self.manager.update(modified_instance, **kwargs)

    def run_and_print(self, args, **kwargs):
        instance = self.run(args, **kwargs)
        try:
            self.print_output(instance, table.PropertyValueTable,
                              attributes=['all'], json=args.json)
        except Exception as e:
            print('ERROR: %s' % e.message)
            raise OperationFailureException(e.message)


class ContentPackResourceUpdateCommand(ResourceUpdateCommand):
    pk_argument_name = 'ref_or_id'


class ResourceEnableCommand(ResourceCommand):
    pk_argument_name = 'name_or_id'

    def __init__(self, resource, *args, **kwargs):
        super(ResourceEnableCommand, self).__init__(resource, 'enable',
            'Enable an existing %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        argument = self.pk_argument_name
        metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
        help = self._get_help_for_argument(resource=resource,

                                           argument=self.pk_argument_name)

        self.parser.add_argument(argument,
                                 metavar=metavar,
                                 help=help)

    @add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        resource_id = getattr(args, self.pk_argument_name, None)
        instance = self.get_resource(resource_id, **kwargs)

        data = instance.serialize()

        if 'ref' in data:
            del data['ref']

        data['enabled'] = True
        modified_instance = self.resource.deserialize(data)

        return self.manager.update(modified_instance, **kwargs)

    def run_and_print(self, args, **kwargs):
        instance = self.run(args, **kwargs)
        self.print_output(instance, table.PropertyValueTable,
                          attributes=['all'], json=args.json)


class ContentPackResourceEnableCommand(ResourceEnableCommand):
    pk_argument_name = 'ref_or_id'


class ResourceDisableCommand(ResourceCommand):
    pk_argument_name = 'name_or_id'

    def __init__(self, resource, *args, **kwargs):
        super(ResourceDisableCommand, self).__init__(resource, 'disable',
            'Disable an existing %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        argument = self.pk_argument_name
        metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
        help = self._get_help_for_argument(resource=resource,

                                           argument=self.pk_argument_name)

        self.parser.add_argument(argument,
                                 metavar=metavar,
                                 help=help)

    @add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        resource_id = getattr(args, self.pk_argument_name, None)
        instance = self.get_resource(resource_id, **kwargs)

        data = instance.serialize()

        if 'ref' in data:
            del data['ref']

        data['enabled'] = False
        modified_instance = self.resource.deserialize(data)

        return self.manager.update(modified_instance, **kwargs)

    def run_and_print(self, args, **kwargs):
        instance = self.run(args, **kwargs)
        self.print_output(instance, table.PropertyValueTable,
                          attributes=['all'], json=args.json)


class ContentPackResourceDisableCommand(ResourceDisableCommand):
    pk_argument_name = 'ref_or_id'


class ResourceDeleteCommand(ResourceCommand):
    pk_argument_name = 'name_or_id'

    def __init__(self, resource, *args, **kwargs):
        super(ResourceDeleteCommand, self).__init__(resource, 'delete',
            'Delete an existing %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        argument = self.pk_argument_name
        metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
        help = self._get_help_for_argument(resource=resource,

                                           argument=self.pk_argument_name)

        self.parser.add_argument(argument,
                                 metavar=metavar,
                                 help=help)

    @add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        resource_id = getattr(args, self.pk_argument_name, None)
        instance = self.get_resource(resource_id, **kwargs)
        self.manager.delete(instance, **kwargs)

    def run_and_print(self, args, **kwargs):
        resource_id = getattr(args, self.pk_argument_name, None)

        try:
            self.run(args, **kwargs)
            print('Resource with id "%s" has been successfully deleted.' % (resource_id))
        except ResourceNotFoundError:
            self.print_not_found(resource_id)
            raise OperationFailureException('Resource %s not found.' % resource_id)


class ContentPackResourceDeleteCommand(ResourceDeleteCommand):
    """
    Base command class for deleting a resource which belongs to a content pack.
    """

    pk_argument_name = 'ref_or_id'


def _load_meta_file(file_path):
    if not os.path.isfile(file_path):
        raise Exception('File "%s" does not exist.' % file_path)

    file_name, file_ext = os.path.splitext(file_path)
    if file_ext not in ALLOWED_EXTS:
        raise Exception('Unsupported meta type %s, file %s. Allowed: %s' %
                        (file_ext, file_path, ALLOWED_EXTS))

    with open(file_path, 'r') as f:
        return PARSER_FUNCS[file_ext](f)


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 os
17			import abc
18			import six
19			import json
20			import logging
21			import httplib
22			from functools import wraps
23
24			import yaml
25
26			from st2client import commands
27			from st2client.exceptions.operations import OperationFailureException
28			from st2client.formatters import table
29
30			ALLOWED_EXTS = ['.json', '.yaml', '.yml']
31			PARSER_FUNCS = {'.json': json.load, '.yml': yaml.safe_load, '.yaml': yaml.safe_load}
32			LOG = logging.getLogger(__name__)
33
34
35			def add_auth_token_to_kwargs_from_cli(func):
36			@wraps(func)
37			def decorate(args, *kwargs):
38			ns = args[1]
39			if getattr(ns, 'token', None):
40			kwargs['token'] = ns.token
41			return func(args, *kwargs)
42			return decorate
43
44
45			class ResourceCommandError(Exception):
46			pass
47
48
49			class ResourceNotFoundError(Exception):
50			pass
51
52
53			class ResourceBranch(commands.Branch):
54
55			def __init__(self, resource, description, app, subparsers,
56			parent_parser=None, read_only=False, commands=None,
			0 ignored issues – show Comprehensibility Bug introduced 2016-02-04 14:00 UTC by Report Bug Copy Issue Report `commands` is re-defining a name which is already available in the outer-scope (previously defined on line `26`). 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...
57			has_disable=False):
58
59			self.resource = resource
60			super(ResourceBranch, self).__init__(
61			self.resource.get_alias().lower(), description,
62			app, subparsers, parent_parser=parent_parser)
63
64			# Registers subcommands for managing the resource type.
65			self.subparsers = self.parser.add_subparsers(
66			help=('List of commands for managing %s.' %
67			self.resource.get_plural_display_name().lower()))
68
69			# Resolves if commands need to be overridden.
70			commands = commands or {}
71			if 'list' not in commands:
72			commands['list'] = ResourceListCommand
73			if 'get' not in commands:
74			commands['get'] = ResourceGetCommand
75			if 'create' not in commands:
76			commands['create'] = ResourceCreateCommand
77			if 'update' not in commands:
78			commands['update'] = ResourceUpdateCommand
79			if 'delete' not in commands:
80			commands['delete'] = ResourceDeleteCommand
81
82			if 'enable' not in commands:
83			commands['enable'] = ResourceEnableCommand
84
85			if 'disable' not in commands:
86			commands['disable'] = ResourceDisableCommand
87
88			# Instantiate commands.
89			args = [self.resource, self.app, self.subparsers]
90			self.commands['list'] = commands['list'](*args)
91			self.commands['get'] = commands['get'](*args)
92
93			if not read_only:
94			self.commands['create'] = commands['create'](*args)
95			self.commands['update'] = commands['update'](*args)
96			self.commands['delete'] = commands['delete'](*args)
97
98			if has_disable:
99			self.commands['enable'] = commands['enable'](*args)
100			self.commands['disable'] = commands['disable'](*args)
101
102
103			@six.add_metaclass(abc.ABCMeta)
104			class ResourceCommand(commands.Command):
105			pk_argument_name = None
106
107			def __init__(self, resource, args, *kwargs):
108
109			has_token_opt = kwargs.pop('has_token_opt', True)
110
111			super(ResourceCommand, self).__init__(args, *kwargs)
112
113			self.resource = resource
114
115			if has_token_opt:
116			self.parser.add_argument('-t', '--token', dest='token',
117			help='Access token for user authentication. '
118			'Get ST2_AUTH_TOKEN from the environment '
119			'variables by default.')
120
121			self.parser.add_argument('-j', '--json',
122			action='store_true', dest='json',
123			help='Prints output in JSON format.')
124
125			@property
126			def manager(self):
127			return self.app.client.managers[self.resource.__name__]
128
129			@property
130			def arg_name_for_resource_id(self):
131			resource_name = self.resource.get_display_name().lower()
132			return '%s-id' % resource_name.replace(' ', '-')
133
134			def print_not_found(self, name):
135			print ('%s "%s" is not found.\n' %
			0 ignored issues – show Coding Style introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report No space allowed before bracket print ('s "s" is not found.\n' % ^ Loading history...
136			(self.resource.get_display_name(), name))
137
138			def get_resource(self, name_or_id, **kwargs):
139			pk_argument_name = self.pk_argument_name
140
141			if pk_argument_name == 'name_or_id':
142			instance = self.get_resource_by_name_or_id(name_or_id=name_or_id, **kwargs)
143			elif pk_argument_name == 'ref_or_id':
144			instance = self.get_resource_by_ref_or_id(ref_or_id=name_or_id, **kwargs)
145			else:
146			instance = self.get_resource_by_pk(pk=name_or_id, **kwargs)
147
148			return instance
149
150			def get_resource_by_pk(self, pk, **kwargs):
151			"""
152			Retrieve resource by a primary key.
153			"""
154			try:
155			instance = self.manager.get_by_id(pk, **kwargs)
156			except Exception as e:
157			# Hack for "Unauthorized" exceptions, we do want to propagate those
158			response = getattr(e, 'response', None)
159			status_code = getattr(response, 'status_code', None)
160			if status_code and status_code == httplib.UNAUTHORIZED:
161			raise e
162
163			instance = None
164
165			return instance
166
167			def get_resource_by_id(self, id, **kwargs):
			0 ignored issues – show Bug Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `id`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
168			instance = self.get_resource_by_pk(pk=id, **kwargs)
169
170			if not instance:
171			message = ('Resource with id "%s" doesn\'t exist.' % (id))
172			raise ResourceNotFoundError(message)
173			return instance
174
175			def get_resource_by_name(self, name, **kwargs):
176			"""
177			Retrieve resource by name.
178			"""
179			instance = self.manager.get_by_name(name, **kwargs)
180			return instance
181
182			def get_resource_by_name_or_id(self, name_or_id, **kwargs):
183			instance = self.get_resource_by_name(name=name_or_id, **kwargs)
184			if not instance:
185			instance = self.get_resource_by_pk(pk=name_or_id, **kwargs)
186
187			if not instance:
188			message = ('Resource with id or name "%s" doesn\'t exist.' %
189			(name_or_id))
190			raise ResourceNotFoundError(message)
191			return instance
192
193			def get_resource_by_ref_or_id(self, ref_or_id, **kwargs):
194			instance = self.manager.get_by_ref_or_id(ref_or_id=ref_or_id, **kwargs)
195
196			if not instance:
197			message = ('Resource with id or reference "%s" doesn\'t exist.' %
198			(ref_or_id))
199			raise ResourceNotFoundError(message)
200			return instance
201
202			@abc.abstractmethod
203			def run(self, args, **kwargs):
204			raise NotImplementedError
205
206			@abc.abstractmethod
207			def run_and_print(self, args, **kwargs):
208			raise NotImplementedError
209
210			def _get_metavar_for_argument(self, argument):
211			return argument.replace('_', '-')
212
213			def _get_help_for_argument(self, resource, argument):
214			argument_display_name = argument.title()
215			resource_display_name = resource.get_display_name().lower()
216
217			if 'ref' in argument:
218			result = ('Reference or ID of the %s.' % (resource_display_name))
219			elif 'name_or_id' in argument:
220			result = ('Name or ID of the %s.' % (resource_display_name))
221			else:
222			result = ('%s of the %s.' % (argument_display_name, resource_display_name))
223
224			return result
225
226
227			class ResourceListCommand(ResourceCommand):
228			display_attributes = ['id', 'name', 'description']
229
230			def __init__(self, resource, args, *kwargs):
231			super(ResourceListCommand, self).__init__(resource, 'list',
232			'Get the list of %s.' % resource.get_plural_display_name().lower(),
233			args, *kwargs)
234
235			self.parser.add_argument('-a', '--attr', nargs='+',
236			default=self.display_attributes,
237			help=('List of attributes to include in the '
238			'output. "all" will return all '
239			'attributes.'))
240			self.parser.add_argument('-w', '--width', nargs='+', type=int,
241			default=None,
242			help=('Set the width of columns in output.'))
243
244			@add_auth_token_to_kwargs_from_cli
245			def run(self, args, **kwargs):
246			return self.manager.get_all(**kwargs)
247
248			def run_and_print(self, args, **kwargs):
249			instances = self.run(args, **kwargs)
250			self.print_output(instances, table.MultiColumnTable,
251			attributes=args.attr, widths=args.width,
252			json=args.json)
253
254
255			class ContentPackResourceListCommand(ResourceListCommand):
256			"""
257			Base command class for use with resources which belong to a content pack.
258			"""
259			def __init__(self, resource, args, *kwargs):
260			super(ContentPackResourceListCommand, self).__init__(resource,
261			args, *kwargs)
262
263			self.parser.add_argument('-p', '--pack', type=str,
264			help=('Only return resources belonging to the'
265			' provided pack'))
266
267			@add_auth_token_to_kwargs_from_cli
268			def run(self, args, **kwargs):
269			filters = {'pack': args.pack}
270			filters.update(**kwargs)
271			return self.manager.get_all(**filters)
272
273
274			class ResourceGetCommand(ResourceCommand):
275			display_attributes = ['all']
276			attribute_display_order = ['id', 'name', 'description']
277
278			pk_argument_name = 'name_or_id' # name of the attribute which stores resource PK
279
280			def __init__(self, resource, args, *kwargs):
281			super(ResourceGetCommand, self).__init__(resource, 'get',
282			'Get individual %s.' % resource.get_display_name().lower(),
283			args, *kwargs)
284
285			argument = self.pk_argument_name
286			metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
287			help = self._get_help_for_argument(resource=resource,
			0 ignored issues – show Bug Best Practice introduced 2016-05-06 11:20 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `help`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
288			argument=self.pk_argument_name)
289
290			self.parser.add_argument(argument,
291			metavar=metavar,
292			help=help)
293			self.parser.add_argument('-a', '--attr', nargs='+',
294			default=self.display_attributes,
295			help=('List of attributes to include in the '
296			'output. "all" or unspecified will '
297			'return all attributes.'))
298
299			@add_auth_token_to_kwargs_from_cli
300			def run(self, args, **kwargs):
301			resource_id = getattr(args, self.pk_argument_name, None)
302			return self.get_resource_by_id(resource_id, **kwargs)
303
304			def run_and_print(self, args, **kwargs):
305			try:
306			instance = self.run(args, **kwargs)
307			self.print_output(instance, table.PropertyValueTable,
308			attributes=args.attr, json=args.json,
309			attribute_display_order=self.attribute_display_order)
310			except ResourceNotFoundError:
311			resource_id = getattr(args, self.pk_argument_name, None)
312			self.print_not_found(resource_id)
313			raise OperationFailureException('Resource %s not found.' % resource_id)
314
315
316			class ContentPackResourceGetCommand(ResourceGetCommand):
317			"""
318			Command for retrieving a single resource which belongs to a content pack.
319
320			Note: All the resources which belong to the content pack can either be
321			retrieved by a reference or by an id.
322			"""
323
324			attribute_display_order = ['id', 'pack', 'name', 'description']
325
326			pk_argument_name = 'ref_or_id'
327
328			def get_resource(self, ref_or_id, **kwargs):
329			return self.get_resource_by_ref_or_id(ref_or_id=ref_or_id, **kwargs)
330
331
332			class ResourceCreateCommand(ResourceCommand):
333
334			def __init__(self, resource, args, *kwargs):
335			super(ResourceCreateCommand, self).__init__(resource, 'create',
336			'Create a new %s.' % resource.get_display_name().lower(),
337			args, *kwargs)
338
339			self.parser.add_argument('file',
340			help=('JSON/YAML file containing the %s to create.'
341			% resource.get_display_name().lower()))
342
343			@add_auth_token_to_kwargs_from_cli
344			def run(self, args, **kwargs):
345			data = _load_meta_file(args.file)
346			instance = self.resource.deserialize(data)
347			return self.manager.create(instance, **kwargs)
348
349			def run_and_print(self, args, **kwargs):
350			try:
351			instance = self.run(args, **kwargs)
352			if not instance:
353			raise Exception('Server did not create instance.')
354			self.print_output(instance, table.PropertyValueTable,
355			attributes=['all'], json=args.json)
356			except Exception as e:
357			message = e.message or str(e)
358			print('ERROR: %s' % (message))
359			raise OperationFailureException(message)
360
361
362			class ResourceUpdateCommand(ResourceCommand):
363			pk_argument_name = 'name_or_id'
364
365			def __init__(self, resource, args, *kwargs):
366			super(ResourceUpdateCommand, self).__init__(resource, 'update',
367			'Updating an existing %s.' % resource.get_display_name().lower(),
368			args, *kwargs)
369
370			argument = self.pk_argument_name
371			metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
372			help = self._get_help_for_argument(resource=resource,
			0 ignored issues – show Bug Best Practice introduced 2016-05-06 11:20 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `help`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
373			argument=self.pk_argument_name)
374
375			self.parser.add_argument(argument,
376			metavar=metavar,
377			help=help)
378			self.parser.add_argument('file',
379			help=('JSON/YAML file containing the %s to update.'
380			% resource.get_display_name().lower()))
381
382			@add_auth_token_to_kwargs_from_cli
383			def run(self, args, **kwargs):
384			resource_id = getattr(args, self.pk_argument_name, None)
385			instance = self.get_resource(resource_id, **kwargs)
386			data = _load_meta_file(args.file)
387			modified_instance = self.resource.deserialize(data)
388
389			if not getattr(modified_instance, 'id', None):
390			modified_instance.id = instance.id
391			else:
392			if modified_instance.id != instance.id:
393			raise Exception('The value for the %s id in the JSON/YAML file '
394			'does not match the ID provided in the '
395			'command line arguments.' %
396			self.resource.get_display_name().lower())
397			return self.manager.update(modified_instance, **kwargs)
398
399			def run_and_print(self, args, **kwargs):
400			instance = self.run(args, **kwargs)
401			try:
402			self.print_output(instance, table.PropertyValueTable,
403			attributes=['all'], json=args.json)
404			except Exception as e:
405			print('ERROR: %s' % e.message)
406			raise OperationFailureException(e.message)
407
408
409			class ContentPackResourceUpdateCommand(ResourceUpdateCommand):
410			pk_argument_name = 'ref_or_id'
411
412
413			class ResourceEnableCommand(ResourceCommand):
414			pk_argument_name = 'name_or_id'
415
416			def __init__(self, resource, args, *kwargs):
417			super(ResourceEnableCommand, self).__init__(resource, 'enable',
418			'Enable an existing %s.' % resource.get_display_name().lower(),
419			args, *kwargs)
420
421			argument = self.pk_argument_name
422			metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
423			help = self._get_help_for_argument(resource=resource,
			0 ignored issues – show Bug Best Practice introduced 2016-05-06 11:20 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `help`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
424			argument=self.pk_argument_name)
425
426			self.parser.add_argument(argument,
427			metavar=metavar,
428			help=help)
429
430			@add_auth_token_to_kwargs_from_cli
431			def run(self, args, **kwargs):
432			resource_id = getattr(args, self.pk_argument_name, None)
433			instance = self.get_resource(resource_id, **kwargs)
434
435			data = instance.serialize()
436
437			if 'ref' in data:
438			del data['ref']
439
440			data['enabled'] = True
441			modified_instance = self.resource.deserialize(data)
442
443			return self.manager.update(modified_instance, **kwargs)
444
445			def run_and_print(self, args, **kwargs):
446			instance = self.run(args, **kwargs)
447			self.print_output(instance, table.PropertyValueTable,
448			attributes=['all'], json=args.json)
449
450
451			class ContentPackResourceEnableCommand(ResourceEnableCommand):
452			pk_argument_name = 'ref_or_id'
453
454
455			class ResourceDisableCommand(ResourceCommand):
456			pk_argument_name = 'name_or_id'
457
458			def __init__(self, resource, args, *kwargs):
459			super(ResourceDisableCommand, self).__init__(resource, 'disable',
460			'Disable an existing %s.' % resource.get_display_name().lower(),
461			args, *kwargs)
462
463			argument = self.pk_argument_name
464			metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
465			help = self._get_help_for_argument(resource=resource,
			0 ignored issues – show Bug Best Practice introduced 2016-05-06 11:20 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `help`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
466			argument=self.pk_argument_name)
467
468			self.parser.add_argument(argument,
469			metavar=metavar,
470			help=help)
471
472			@add_auth_token_to_kwargs_from_cli
473			def run(self, args, **kwargs):
474			resource_id = getattr(args, self.pk_argument_name, None)
475			instance = self.get_resource(resource_id, **kwargs)
476
477			data = instance.serialize()
478
479			if 'ref' in data:
480			del data['ref']
481
482			data['enabled'] = False
483			modified_instance = self.resource.deserialize(data)
484
485			return self.manager.update(modified_instance, **kwargs)
486
487			def run_and_print(self, args, **kwargs):
488			instance = self.run(args, **kwargs)
489			self.print_output(instance, table.PropertyValueTable,
490			attributes=['all'], json=args.json)
491
492
493			class ContentPackResourceDisableCommand(ResourceDisableCommand):
494			pk_argument_name = 'ref_or_id'
495
496
497			class ResourceDeleteCommand(ResourceCommand):
498			pk_argument_name = 'name_or_id'
499
500			def __init__(self, resource, args, *kwargs):
501			super(ResourceDeleteCommand, self).__init__(resource, 'delete',
502			'Delete an existing %s.' % resource.get_display_name().lower(),
503			args, *kwargs)
504
505			argument = self.pk_argument_name
506			metavar = self._get_metavar_for_argument(argument=self.pk_argument_name)
507			help = self._get_help_for_argument(resource=resource,
			0 ignored issues – show Bug Best Practice introduced 2016-05-06 11:20 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `help`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
508			argument=self.pk_argument_name)
509
510			self.parser.add_argument(argument,
511			metavar=metavar,
512			help=help)
513
514			@add_auth_token_to_kwargs_from_cli
515			def run(self, args, **kwargs):
516			resource_id = getattr(args, self.pk_argument_name, None)
517			instance = self.get_resource(resource_id, **kwargs)
518			self.manager.delete(instance, **kwargs)
519
520			def run_and_print(self, args, **kwargs):
521			resource_id = getattr(args, self.pk_argument_name, None)
522
523			try:
524			self.run(args, **kwargs)
525			print('Resource with id "%s" has been successfully deleted.' % (resource_id))
526			except ResourceNotFoundError:
527			self.print_not_found(resource_id)
528			raise OperationFailureException('Resource %s not found.' % resource_id)
529
530
531			class ContentPackResourceDeleteCommand(ResourceDeleteCommand):
532			"""
533			Base command class for deleting a resource which belongs to a content pack.
534			"""
535
536			pk_argument_name = 'ref_or_id'
537
538
539			def _load_meta_file(file_path):
540			if not os.path.isfile(file_path):
541			raise Exception('File "%s" does not exist.' % file_path)
542
543			file_name, file_ext = os.path.splitext(file_path)
544			if file_ext not in ALLOWED_EXTS:
545			raise Exception('Unsupported meta type %s, file %s. Allowed: %s' %
546			(file_ext, file_path, ALLOWED_EXTS))
547
548			with open(file_path, 'r') as f:
549			return PARSER_FUNCS[file_ext](f)
550

StackStorm / st2

Push — master ( 58fbab...4c05e2 )

ResourceBranch.__init__() F

Complexity

Size

Duplication

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

ResourceBranch.init() F