ApiKeyListCommand.run_and_print() - 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

ApiKeyListCommand.run_and_print() A

↳ Parent: ApiKeyListCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
dl	0
loc	6
rs	9.4285
c	1
b	0
f	0

# 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 getpass
import json
import logging

from st2client import models
from st2client.commands import resource
from st2client.commands.noop import NoopCommand
from st2client.exceptions.operations import OperationFailureException
from st2client.formatters import table


LOG = logging.getLogger(__name__)


class TokenCreateCommand(resource.ResourceCommand):

    display_attributes = ['user', 'token', 'expiry']

    def __init__(self, resource, *args, **kwargs):
param = 5

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

        kwargs['has_token_opt'] = False

        super(TokenCreateCommand, self).__init__(
            resource, kwargs.pop('name', 'create'),
            'Authenticate user and aquire access token.',
            *args, **kwargs)

        self.parser.add_argument('username',
                                 help='Name of the user to authenticate.')

        self.parser.add_argument('-p', '--password', dest='password',
                                 help='Password for the user. If password is not provided, '
                                      'it will be prompted.')
        self.parser.add_argument('-l', '--ttl', type=int, dest='ttl', default=None,
                                 help='The life span of the token in seconds. '
                                      'Max TTL configured by the admin supersedes this.')
        self.parser.add_argument('-t', '--only-token', action='store_true', dest='only_token',
                                 default=False,
                                 help='Only print token to the console on successful '
                                      'authentication.')

    def run(self, args, **kwargs):
        if not args.password:
            args.password = getpass.getpass()
        instance = self.resource(ttl=args.ttl) if args.ttl else self.resource()
        return self.manager.create(instance, auth=(args.username, args.password), **kwargs)

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

        if args.only_token:
            print(instance.token)
        else:
            self.print_output(instance, table.PropertyValueTable,
                              attributes=self.display_attributes, json=args.json, yaml=args.yaml)


class ApiKeyBranch(resource.ResourceBranch):

    def __init__(self, description, app, subparsers, parent_parser=None):
        super(ApiKeyBranch, self).__init__(
            models.ApiKey, description, app, subparsers,
            parent_parser=parent_parser,
            commands={
                'list': ApiKeyListCommand,
                'get': ApiKeyGetCommand,
                'create': ApiKeyCreateCommand,
                'update': NoopCommand,
                'delete': ApiKeyDeleteCommand
            })

        self.commands['enable'] = ApiKeyEnableCommand(self.resource, self.app, self.subparsers)
        self.commands['disable'] = ApiKeyDisableCommand(self.resource, self.app, self.subparsers)
        self.commands['load'] = ApiKeyLoadCommand(self.resource, self.app, self.subparsers)


class ApiKeyListCommand(resource.ResourceListCommand):
    detail_display_attributes = ['all']
    display_attributes = ['id', 'user', 'metadata']

    def __init__(self, resource, *args, **kwargs):
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(ApiKeyListCommand, self).__init__(resource, *args, **kwargs)

        self.parser.add_argument('-u', '--user', type=str,
                                 help='Only return ApiKeys belonging to the provided user')
        self.parser.add_argument('-d', '--detail', action='store_true',
                                 help='Full list of attributes.')

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

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


class ApiKeyGetCommand(resource.ResourceGetCommand):
    display_attributes = ['all']
    attribute_display_order = ['id', 'user', 'metadata']

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


class ApiKeyCreateCommand(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(ApiKeyCreateCommand, self).__init__(
            resource, 'create', 'Create a new %s.' % resource.get_display_name().lower(),
            *args, **kwargs)

        self.parser.add_argument('-u', '--user', type=str,
                                 help='User for which to create API Keys.',
                                 default='')
        self.parser.add_argument('-m', '--metadata', type=json.loads,
                                 help='User for which to create API Keys.',
                                 default={})
        self.parser.add_argument('-k', '--only-key', action='store_true', dest='only_key',
                                 default=False,
                                 help='Only print API Key to the console on creation.')

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        data = {}
        if args.user:
            data['user'] = args.user
        if args.metadata:
            data['metadata'] = args.metadata
        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.')
        except Exception as e:
            message = e.message or str(e)
            print('ERROR: %s' % (message))
            raise OperationFailureException(message)
        if args.only_key:
            print(instance.key)
        else:
            self.print_output(instance, table.PropertyValueTable,
                              attributes=['all'], json=args.json, yaml=args.yaml)


class ApiKeyLoadCommand(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(ApiKeyLoadCommand, self).__init__(
            resource, 'load', 'Load %s from a file.' % resource.get_display_name().lower(),
            *args, **kwargs)

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

        self.parser.add_argument('-w', '--width', nargs='+', type=int,
                                 default=None,
                                 help=('Set the width of columns in output.'))

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        resources = resource.load_meta_file(args.file)
        if not resources:
            print('No %s found in %s.' % (self.resource.get_display_name().lower(), args.file))
            return None
        if not isinstance(resources, list):
            resources = [resources]
        instances = []
        for res in resources:
            # pick only the meaningful properties.
            instance = {
                'user': res['user'],  # required
                'key_hash': res['key_hash'],  # required
                'metadata': res.get('metadata', {}),
                'enabled': res.get('enabled', False)
            }
            instance = self.resource.deserialize(instance)
            instances.append(self.manager.create(instance, **kwargs))
        return instances

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


class ApiKeyDeleteCommand(resource.ResourceDeleteCommand):
    pk_argument_name = 'key_or_id'  # name of the attribute which stores resource PK


class ApiKeyEnableCommand(resource.ResourceEnableCommand):
    pk_argument_name = 'key_or_id'  # name of the attribute which stores resource PK


class ApiKeyDisableCommand(resource.ResourceDisableCommand):
    pk_argument_name = 'key_or_id'  # name of the attribute which stores resource PK


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 getpass
17			import json
18			import logging
19
20			from st2client import models
21			from st2client.commands import resource
22			from st2client.commands.noop import NoopCommand
23			from st2client.exceptions.operations import OperationFailureException
24			from st2client.formatters import table
25
26
27			LOG = logging.getLogger(__name__)
28
29
30			class TokenCreateCommand(resource.ResourceCommand):
31
32			display_attributes = ['user', 'token', 'expiry']
33
34			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-02-04 14:00 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 `21`). 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...
35
36			kwargs['has_token_opt'] = False
37
38			super(TokenCreateCommand, self).__init__(
39			resource, kwargs.pop('name', 'create'),
40			'Authenticate user and aquire access token.',
41			args, *kwargs)
42
43			self.parser.add_argument('username',
44			help='Name of the user to authenticate.')
45
46			self.parser.add_argument('-p', '--password', dest='password',
47			help='Password for the user. If password is not provided, '
48			'it will be prompted.')
49			self.parser.add_argument('-l', '--ttl', type=int, dest='ttl', default=None,
50			help='The life span of the token in seconds. '
51			'Max TTL configured by the admin supersedes this.')
52			self.parser.add_argument('-t', '--only-token', action='store_true', dest='only_token',
53			default=False,
54			help='Only print token to the console on successful '
55			'authentication.')
56
57			def run(self, args, **kwargs):
58			if not args.password:
59			args.password = getpass.getpass()
60			instance = self.resource(ttl=args.ttl) if args.ttl else self.resource()
61			return self.manager.create(instance, auth=(args.username, args.password), **kwargs)
62
63			def run_and_print(self, args, **kwargs):
64			instance = self.run(args, **kwargs)
65
66			if args.only_token:
67			print(instance.token)
68			else:
69			self.print_output(instance, table.PropertyValueTable,
70			attributes=self.display_attributes, json=args.json, yaml=args.yaml)
71
72
73			class ApiKeyBranch(resource.ResourceBranch):
74
75			def __init__(self, description, app, subparsers, parent_parser=None):
76			super(ApiKeyBranch, self).__init__(
77			models.ApiKey, description, app, subparsers,
78			parent_parser=parent_parser,
79			commands={
80			'list': ApiKeyListCommand,
81			'get': ApiKeyGetCommand,
82			'create': ApiKeyCreateCommand,
83			'update': NoopCommand,
84			'delete': ApiKeyDeleteCommand
85			})
86
87			self.commands['enable'] = ApiKeyEnableCommand(self.resource, self.app, self.subparsers)
88			self.commands['disable'] = ApiKeyDisableCommand(self.resource, self.app, self.subparsers)
89			self.commands['load'] = ApiKeyLoadCommand(self.resource, self.app, self.subparsers)
90
91
92			class ApiKeyListCommand(resource.ResourceListCommand):
93			detail_display_attributes = ['all']
94			display_attributes = ['id', 'user', 'metadata']
95
96			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-02-04 14:00 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 `21`). 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...
97			super(ApiKeyListCommand, self).__init__(resource, args, *kwargs)
98
99			self.parser.add_argument('-u', '--user', type=str,
100			help='Only return ApiKeys belonging to the provided user')
101			self.parser.add_argument('-d', '--detail', action='store_true',
102			help='Full list of attributes.')
103
104			@resource.add_auth_token_to_kwargs_from_cli
105			def run(self, args, **kwargs):
106			filters = {}
107			filters['user'] = args.user
108			filters.update(**kwargs)
109			return self.manager.get_all(**filters)
110
111			def run_and_print(self, args, **kwargs):
112			instances = self.run(args, **kwargs)
113			attr = self.detail_display_attributes if args.detail else args.attr
114			self.print_output(instances, table.MultiColumnTable,
115			attributes=attr, widths=args.width,
116			json=args.json, yaml=args.yaml)
117
118
119			class ApiKeyGetCommand(resource.ResourceGetCommand):
120			display_attributes = ['all']
121			attribute_display_order = ['id', 'user', 'metadata']
122
123			pk_argument_name = 'key_or_id' # name of the attribute which stores resource PK
124
125
126			class ApiKeyCreateCommand(resource.ResourceCommand):
127
128			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-02-04 14:00 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 `21`). 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...
129			super(ApiKeyCreateCommand, self).__init__(
130			resource, 'create', 'Create a new %s.' % resource.get_display_name().lower(),
131			args, *kwargs)
132
133			self.parser.add_argument('-u', '--user', type=str,
134			help='User for which to create API Keys.',
135			default='')
136			self.parser.add_argument('-m', '--metadata', type=json.loads,
137			help='User for which to create API Keys.',
138			default={})
139			self.parser.add_argument('-k', '--only-key', action='store_true', dest='only_key',
140			default=False,
141			help='Only print API Key to the console on creation.')
142
143			@resource.add_auth_token_to_kwargs_from_cli
144			def run(self, args, **kwargs):
145			data = {}
146			if args.user:
147			data['user'] = args.user
148			if args.metadata:
149			data['metadata'] = args.metadata
150			instance = self.resource.deserialize(data)
151			return self.manager.create(instance, **kwargs)
152
153			def run_and_print(self, args, **kwargs):
154			try:
155			instance = self.run(args, **kwargs)
156			if not instance:
157			raise Exception('Server did not create instance.')
158			except Exception as e:
159			message = e.message or str(e)
160			print('ERROR: %s' % (message))
161			raise OperationFailureException(message)
162			if args.only_key:
163			print(instance.key)
164			else:
165			self.print_output(instance, table.PropertyValueTable,
166			attributes=['all'], json=args.json, yaml=args.yaml)
167
168
169			class ApiKeyLoadCommand(resource.ResourceCommand):
170
171			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2016-05-24 01:09 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 `21`). 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...
172			super(ApiKeyLoadCommand, self).__init__(
173			resource, 'load', 'Load %s from a file.' % resource.get_display_name().lower(),
174			args, *kwargs)
175
176			self.parser.add_argument('file',
177			help=('JSON/YAML file containing the %s(s) to load.'
178			% resource.get_display_name().lower()),
179			default='')
180
181			self.parser.add_argument('-w', '--width', nargs='+', type=int,
182			default=None,
183			help=('Set the width of columns in output.'))
184
185			@resource.add_auth_token_to_kwargs_from_cli
186			def run(self, args, **kwargs):
187			resources = resource.load_meta_file(args.file)
188			if not resources:
189			print('No %s found in %s.' % (self.resource.get_display_name().lower(), args.file))
190			return None
191			if not isinstance(resources, list):
192			resources = [resources]
193			instances = []
194			for res in resources:
195			# pick only the meaningful properties.
196			instance = {
197			'user': res['user'], # required
198			'key_hash': res['key_hash'], # required
199			'metadata': res.get('metadata', {}),
200			'enabled': res.get('enabled', False)
201			}
202			instance = self.resource.deserialize(instance)
203			instances.append(self.manager.create(instance, **kwargs))
204			return instances
205
206			def run_and_print(self, args, **kwargs):
207			instances = self.run(args, **kwargs)
208			if instances:
209			self.print_output(instances, table.MultiColumnTable,
210			attributes=ApiKeyListCommand.display_attributes,
211			widths=args.width,
212			json=args.json, yaml=args.yaml)
213
214
215			class ApiKeyDeleteCommand(resource.ResourceDeleteCommand):
216			pk_argument_name = 'key_or_id' # name of the attribute which stores resource PK
217
218
219			class ApiKeyEnableCommand(resource.ResourceEnableCommand):
220			pk_argument_name = 'key_or_id' # name of the attribute which stores resource PK
221
222
223			class ApiKeyDisableCommand(resource.ResourceDisableCommand):
224			pk_argument_name = 'key_or_id' # name of the attribute which stores resource PK
225

StackStorm / st2

Pull Request — master (#2710)

ApiKeyListCommand.run_and_print() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like