TriggerInstanceListCommand.run() - Code Metrics - Inspection of "[WIP] Ti msg ack" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2643)

by Manas

created 2016-04-27 00:42 UTC

TriggerInstanceListCommand.run() B

↳ Parent: TriggerInstanceListCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
c	1
b	0
f	0
dl	0
loc	13
rs	8.5454

# 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.

from st2client.commands import resource
from st2client.formatters import table
from st2client.models import TriggerInstance
from st2client.utils.date import format_isodate_for_user_timezone


class TriggerInstanceResendCommand(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(TriggerInstanceResendCommand, self).__init__(
            resource, kwargs.pop('name', 're-emit'),
            'A command to re-emit a particular trigger instance.',
            *args, **kwargs)

        self.parser.add_argument('id', nargs='?',
                                 metavar='id',
                                 help='ID of trigger instance to re-emit.')
        self.parser.add_argument(
            '-h', '--help',
            action='store_true', dest='help',
            help='Print usage for the given command.')

    def run(self, args, **kwargs):
        return self.manager.re_emit(args.id)

    @resource.add_auth_token_to_kwargs_from_cli
    def run_and_print(self, args, **kwargs):
        ret = self.run(args, **kwargs)
        if 'message' in ret:
            print(ret['message'])


class TriggerInstanceBranch(resource.ResourceBranch):
    def __init__(self, description, app, subparsers, parent_parser=None):
        super(TriggerInstanceBranch, self).__init__(
            TriggerInstance, description, app, subparsers,
            parent_parser=parent_parser, read_only=True,
            commands={
                'list': TriggerInstanceListCommand,
                'get': TriggerInstanceGetCommand
            })

        self.commands['re-emit'] = TriggerInstanceResendCommand(self.resource, self.app,
                                                                self.subparsers, add_help=False)


class TriggerInstanceListCommand(resource.ResourceCommand):
    display_attributes = ['id', 'trigger', 'occurrence_time', 'status']

    attribute_transform_functions = {
        'occurrence_time': format_isodate_for_user_timezone
    }

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

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        super(TriggerInstanceListCommand, self).__init__(
            resource, 'list', 'Get the list of the 50 most recent %s.' %
            resource.get_plural_display_name().lower(),
            *args, **kwargs)

        self.group = self.parser.add_argument_group()
        self.parser.add_argument('-n', '--last', type=int, dest='last',
                                 default=50,
                                 help=('List N most recent %s; '
                                       'list all if 0.' %
                                       resource.get_plural_display_name().lower()))

        # Filter options
        self.group.add_argument('--trigger', help='Trigger reference to filter the list.')

        self.parser.add_argument('-tg', '--timestamp-gt', type=str, dest='timestamp_gt',
                                 default=None,
                                 help=('Only return trigger instances with occurrence_time '
                                       'greater than the one provided. '
                                       'Use time in the format 2000-01-01T12:00:00.000Z'))
        self.parser.add_argument('-tl', '--timestamp-lt', type=str, dest='timestamp_lt',
                                 default=None,
                                 help=('Only return trigger instances with timestamp '
                                       'lower than the one provided. '
                                       'Use time in the format 2000-01-01T12:00:00.000Z'))

        self.group.add_argument('--status',
                                help='Can be pending, processing, processed or processing_failed.')

        # Display options
        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.'))

    @resource.add_auth_token_to_kwargs_from_cli
    def run(self, args, **kwargs):
        # Filtering options
        if args.trigger:
            kwargs['trigger'] = args.trigger
        if args.timestamp_gt:
            kwargs['timestamp_gt'] = args.timestamp_gt
        if args.timestamp_lt:
            kwargs['timestamp_lt'] = args.timestamp_lt
        if args.status:
            kwargs['status'] = args.status

        return self.manager.query(limit=args.last, **kwargs)

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


class TriggerInstanceGetCommand(resource.ResourceGetCommand):
    display_attributes = ['all']
    attribute_display_order = ['id', 'trigger', 'occurrence_time', 'payload']

    pk_argument_name = 'id'

    @resource.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)


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			from st2client.commands import resource
17			from st2client.formatters import table
18			from st2client.models import TriggerInstance
19			from st2client.utils.date import format_isodate_for_user_timezone
20
21
22			class TriggerInstanceResendCommand(resource.ResourceCommand):
23			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 `16`). 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...
24
25			super(TriggerInstanceResendCommand, self).__init__(
26			resource, kwargs.pop('name', 're-emit'),
27			'A command to re-emit a particular trigger instance.',
28			args, *kwargs)
29
30			self.parser.add_argument('id', nargs='?',
31			metavar='id',
32			help='ID of trigger instance to re-emit.')
33			self.parser.add_argument(
34			'-h', '--help',
35			action='store_true', dest='help',
36			help='Print usage for the given command.')
37
38			def run(self, args, **kwargs):
39			return self.manager.re_emit(args.id)
40
41			@resource.add_auth_token_to_kwargs_from_cli
42			def run_and_print(self, args, **kwargs):
43			ret = self.run(args, **kwargs)
44			if 'message' in ret:
45			print(ret['message'])
46
47
48			class TriggerInstanceBranch(resource.ResourceBranch):
49			def __init__(self, description, app, subparsers, parent_parser=None):
50			super(TriggerInstanceBranch, self).__init__(
51			TriggerInstance, description, app, subparsers,
52			parent_parser=parent_parser, read_only=True,
53			commands={
54			'list': TriggerInstanceListCommand,
55			'get': TriggerInstanceGetCommand
56			})
57
58			self.commands['re-emit'] = TriggerInstanceResendCommand(self.resource, self.app,
59			self.subparsers, add_help=False)
60
61
62			class TriggerInstanceListCommand(resource.ResourceCommand):
63			display_attributes = ['id', 'trigger', 'occurrence_time', 'status']
64
65			attribute_transform_functions = {
66			'occurrence_time': format_isodate_for_user_timezone
67			}
68
69			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 `16`). 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...
70			super(TriggerInstanceListCommand, self).__init__(
71			resource, 'list', 'Get the list of the 50 most recent %s.' %
72			resource.get_plural_display_name().lower(),
73			args, *kwargs)
74
75			self.group = self.parser.add_argument_group()
76			self.parser.add_argument('-n', '--last', type=int, dest='last',
77			default=50,
78			help=('List N most recent %s; '
79			'list all if 0.' %
80			resource.get_plural_display_name().lower()))
81
82			# Filter options
83			self.group.add_argument('--trigger', help='Trigger reference to filter the list.')
84
85			self.parser.add_argument('-tg', '--timestamp-gt', type=str, dest='timestamp_gt',
86			default=None,
87			help=('Only return trigger instances with occurrence_time '
88			'greater than the one provided. '
89			'Use time in the format 2000-01-01T12:00:00.000Z'))
90			self.parser.add_argument('-tl', '--timestamp-lt', type=str, dest='timestamp_lt',
91			default=None,
92			help=('Only return trigger instances with timestamp '
93			'lower than the one provided. '
94			'Use time in the format 2000-01-01T12:00:00.000Z'))
95
96			self.group.add_argument('--status',
97			help='Can be pending, processing, processed or processing_failed.')
98
99			# Display options
100			self.parser.add_argument('-a', '--attr', nargs='+',
101			default=self.display_attributes,
102			help=('List of attributes to include in the '
103			'output. "all" will return all '
104			'attributes.'))
105			self.parser.add_argument('-w', '--width', nargs='+', type=int,
106			default=None,
107			help=('Set the width of columns in output.'))
108
109			@resource.add_auth_token_to_kwargs_from_cli
110			def run(self, args, **kwargs):
111			# Filtering options
112			if args.trigger:
113			kwargs['trigger'] = args.trigger
114			if args.timestamp_gt:
115			kwargs['timestamp_gt'] = args.timestamp_gt
116			if args.timestamp_lt:
117			kwargs['timestamp_lt'] = args.timestamp_lt
118			if args.status:
119			kwargs['status'] = args.status
120
121			return self.manager.query(limit=args.last, **kwargs)
122
123			def run_and_print(self, args, **kwargs):
124			instances = self.run(args, **kwargs)
125			self.print_output(reversed(instances), table.MultiColumnTable,
126			attributes=args.attr, widths=args.width,
127			json=args.json,
128			attribute_transform_functions=self.attribute_transform_functions)
129
130
131			class TriggerInstanceGetCommand(resource.ResourceGetCommand):
132			display_attributes = ['all']
133			attribute_display_order = ['id', 'trigger', 'occurrence_time', 'payload']
134
135			pk_argument_name = 'id'
136
137			@resource.add_auth_token_to_kwargs_from_cli
138			def run(self, args, **kwargs):
139			resource_id = getattr(args, self.pk_argument_name, None)
140			return self.get_resource_by_id(resource_id, **kwargs)
141

StackStorm / st2

Pull Request — master (#2643)

TriggerInstanceListCommand.run() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like