_load_meta_file() - Code Metrics - Inspection of "[RFR] Ability to load apikeys from a file." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2710)

by Manas

created 2016-05-24 17:40 UTC

_load_meta_file() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	4
dl	0
loc	11
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 os
import abc
import six
import json
import logging
import httplib
from functools import wraps
import traceback

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.')

        # Formatter flags
        self.parser.add_argument('-j', '--json',
                                 action='store_true', dest='json',
                                 help='Prints output in JSON format.')
        self.parser.add_argument('-y', '--yaml',
                                 action='store_true', dest='yaml',
                                 help='Prints output in YAML 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:
            traceback.print_exc()
            # 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, yaml=args.yaml)


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, yaml=args.yaml,
                              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, yaml=args.yaml)
        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, yaml=args.yaml)
        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, yaml=args.yaml)


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, yaml=args.yaml)


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

StackStorm / st2

Pull Request — master (#2710)

_load_meta_file() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like