TraceListCommand.__init__() - Code Metrics - Inspection of "Fix wrong information for /api/v1/apikeys/ doc" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#4194)

by Arma

created 2018-06-24 10:28 UTC

TraceListCommand.init() A

↳ Parent: TraceListCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
dl	0
loc	36
rs	9.016
c	0
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.

from __future__ import absolute_import

from st2client.models import Resource, Trace, TriggerInstance, Rule, LiveAction
from st2client.exceptions.operations import OperationFailureException
from st2client.formatters import table
from st2client.formatters import execution as execution_formatter
from st2client.commands import resource
from st2client.utils.date import format_isodate_for_user_timezone


TRACE_ATTRIBUTE_DISPLAY_ORDER = ['id', 'trace_tag', 'action_executions', 'rules',
                                 'trigger_instances', 'start_timestamp']

TRACE_HEADER_DISPLAY_ORDER = ['id', 'trace_tag', 'start_timestamp']

TRACE_COMPONENT_DISPLAY_LABELS = ['id', 'type', 'ref', 'updated_at']

TRACE_DISPLAY_ATTRIBUTES = ['all']

TRIGGER_INSTANCE_DISPLAY_OPTIONS = [
    'all',
    'trigger-instances',
    'trigger_instances',
    'triggerinstances',
    'triggers'
]

ACTION_EXECUTION_DISPLAY_OPTIONS = [
    'all',
    'executions',
    'action-executions',
    'action_executions',
    'actionexecutions',
    'actions'
]


class TraceBranch(resource.ResourceBranch):
    def __init__(self, description, app, subparsers, parent_parser=None):
        super(TraceBranch, self).__init__(
            Trace, description, app, subparsers,
            parent_parser=parent_parser,
            read_only=True,
            commands={
                'list': TraceListCommand,
                'get': TraceGetCommand
            })


class SingleTraceDisplayMixin(object):

    def print_trace_details(self, trace, args, **kwargs):
        options = {'attributes': TRACE_ATTRIBUTE_DISPLAY_ORDER if args.json else
                   TRACE_HEADER_DISPLAY_ORDER}
        options['json'] = args.json
        options['yaml'] = args.yaml
        options['attribute_transform_functions'] = self.attribute_transform_functions

        formatter = execution_formatter.ExecutionResult

        self.print_output(trace, formatter, **options)

        # Everything should be printed if we are printing json.
        if args.json or args.yaml:
            return

        components = []
        if any(attr in args.attr for attr in TRIGGER_INSTANCE_DISPLAY_OPTIONS):
            components.extend([Resource(**{'id': trigger_instance['object_id'],
                                           'type': TriggerInstance._alias.lower(),
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                                           'ref': trigger_instance['ref'],
                                           'updated_at': trigger_instance['updated_at']})
                               for trigger_instance in trace.trigger_instances])
        if any(attr in args.attr for attr in ['all', 'rules']):
            components.extend([Resource(**{'id': rule['object_id'],
                                           'type': Rule._alias.lower(),
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                                           'ref': rule['ref'],
                                           'updated_at': rule['updated_at']})
                               for rule in trace.rules])
        if any(attr in args.attr for attr in ACTION_EXECUTION_DISPLAY_OPTIONS):
            components.extend([Resource(**{'id': execution['object_id'],
                                           'type': LiveAction._alias.lower(),
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                                           'ref': execution['ref'],
                                           'updated_at': execution['updated_at']})
                               for execution in trace.action_executions])
        if components:
            components.sort(key=lambda resource: resource.updated_at)
            self.print_output(components, table.MultiColumnTable,
                              attributes=TRACE_COMPONENT_DISPLAY_LABELS,
                              json=args.json, yaml=args.yaml)


class TraceListCommand(resource.ResourceCommand, SingleTraceDisplayMixin):
    display_attributes = ['id', 'uid', 'trace_tag', 'start_timestamp']

    attribute_transform_functions = {
        'start_timestamp': format_isodate_for_user_timezone
    }

    attribute_display_order = TRACE_ATTRIBUTE_DISPLAY_ORDER

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

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

        self.default_limit = 50

        super(TraceListCommand, self).__init__(
            resource, 'list', 'Get the list of the %s most recent %s.' %
            (self.default_limit, resource.get_plural_display_name().lower()),
            *args, **kwargs)

        self.resource_name = resource.get_plural_display_name().lower()
        self.group = self.parser.add_mutually_exclusive_group()
        self.parser.add_argument('-n', '--last', type=int, dest='last',
                                 default=self.default_limit,
                                 help=('List N most recent %s. Use -n -1 to fetch the full result \
                                       set.' % self.resource_name))
        self.parser.add_argument('-s', '--sort', type=str, dest='sort_order',
                                 default='descending',
                                 help=('Sort %s by start timestamp, '
                                       'asc|ascending (earliest first) '
                                       'or desc|descending (latest first)' % self.resource_name))

        # Filter options
        self.group.add_argument('-c', '--trace-tag', help='Trace-tag to filter the list.')
        self.group.add_argument('-e', '--execution', help='Execution to filter the list.')
        self.group.add_argument('-r', '--rule', help='Rule to filter the list.')
        self.group.add_argument('-g', '--trigger-instance',
                                help='TriggerInstance to filter the list.')
        # 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.trace_tag:
            kwargs['trace_tag'] = args.trace_tag
        if args.trigger_instance:
            kwargs['trigger_instance'] = args.trigger_instance
        if args.execution:
            kwargs['execution'] = args.execution
        if args.rule:
            kwargs['rule'] = args.rule

        if args.sort_order:
            if args.sort_order in ['asc', 'ascending']:
                kwargs['sort_asc'] = True
            elif args.sort_order in ['desc', 'descending']:
                kwargs['sort_desc'] = True
        return self.manager.query_with_count(limit=args.last, **kwargs)

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

        if instances and len(instances) == 1:
            # For a single Trace we must include the components unless
            # user has overriden the attributes to display
            if args.attr == self.display_attributes:
                args.attr = ['all']
            self.print_trace_details(trace=instances[0], args=args)

            if not args.json and not args.yaml:
                if args.last and count and count > args.last:
                        table.SingleRowTable.note_box(self.resource_name, 1)

        else:
            if args.json or args.yaml:
                self.print_output(instances, table.MultiColumnTable,
                                  attributes=args.attr, widths=args.width,
                                  json=args.json, yaml=args.yaml,
                                  attribute_transform_functions=self.attribute_transform_functions)
            else:
                self.print_output(instances, table.MultiColumnTable,
                                  attributes=args.attr, widths=args.width,
                                  attribute_transform_functions=self.attribute_transform_functions)

                if args.last and count and count > args.last:
                    table.SingleRowTable.note_box(self.resource_name, args.last)


class TraceGetCommand(resource.ResourceGetCommand, SingleTraceDisplayMixin):
    display_attributes = ['all']
    attribute_display_order = TRACE_ATTRIBUTE_DISPLAY_ORDER
    attribute_transform_functions = {
        'start_timestamp': format_isodate_for_user_timezone
    }

    pk_argument_name = 'id'

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

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

        # Causation chains
        self.causation_group = self.parser.add_mutually_exclusive_group()

        self.causation_group.add_argument('-e', '--execution',
                                          help='Execution to show causation chain.')
        self.causation_group.add_argument('-r', '--rule', help='Rule to show causation chain.')
        self.causation_group.add_argument('-g', '--trigger-instance',
                                          help='TriggerInstance to show causation chain.')

        # display filter group
        self.display_filter_group = self.parser.add_argument_group()

        self.display_filter_group.add_argument('--show-executions', action='store_true',
                                               help='Only show executions.')
        self.display_filter_group.add_argument('--show-rules', action='store_true',
                                               help='Only show rules.')
        self.display_filter_group.add_argument('--show-trigger-instances', action='store_true',
                                               help='Only show trigger instances.')
        self.display_filter_group.add_argument('-n', '--hide-noop-triggers', action='store_true',
                                               help='Hide noop trigger instances.')

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

    @resource.add_auth_token_to_kwargs_from_cli
    def run_and_print(self, args, **kwargs):
        trace = None
        try:
            trace = self.run(args, **kwargs)
        except resource.ResourceNotFoundError:
            self.print_not_found(args.id)
            raise OperationFailureException('Trace %s not found.' % (args.id))
        # First filter for causation chains
        trace = self._filter_trace_components(trace=trace, args=args)
        # next filter for display purposes
        trace = self._apply_display_filters(trace=trace, args=args)
        return self.print_trace_details(trace=trace, args=args)

    @staticmethod
    def _filter_trace_components(trace, args):
        """
        This function walks up the component causal chain. It only returns
        properties in the causal chain and nothing else.
        """
        # check if any filtering is desired
        if not (args.execution or args.rule or args.trigger_instance):
            return trace

        component_id = None
        component_type = None

        # pick the right component type
        if args.execution:
            component_id = args.execution
            component_type = 'action_execution'
        elif args.rule:
            component_id = args.rule
            component_type = 'rule'
        elif args.trigger_instance:
            component_id = args.trigger_instance
            component_type = 'trigger_instance'

        # Initialize collection to use
        action_executions = []
        rules = []
        trigger_instances = []

        # setup flag to properly manage termination conditions
        search_target_found = component_id and component_type

        while search_target_found:
            components_list = []
            if component_type == 'action_execution':
                components_list = trace.action_executions
                to_update_list = action_executions
            elif component_type == 'rule':
                components_list = trace.rules
                to_update_list = rules
            elif component_type == 'trigger_instance':
                components_list = trace.trigger_instances
                to_update_list = trigger_instances
            # Look for search_target in the right collection and
            # once found look up the caused_by to keep movig up
            # the chain.
            search_target_found = False
            # init to default value
            component_caused_by_id = None
            for component in components_list:
                test_id = component['object_id']
                if test_id == component_id:
                    caused_by = component.get('caused_by', {})
                    component_id = caused_by.get('id', None)
                    component_type = caused_by.get('type', None)
                    # If provided the component_caused_by_id must match as well. This is mostly
                    # applicable for rules since the same rule may appear multiple times and can
                    # only be distinguished by causing TriggerInstance.
                    if component_caused_by_id and component_caused_by_id != component_id:
                        continue
                    component_caused_by_id = None
                    to_update_list.append(component)
                    # In some cases the component_id and the causing component are combined to
                    # provide the complete causation chain. Think rule + triggerinstance
                    if component_id and ':' in component_id:
                        component_id_split = component_id.split(':')
                        component_id = component_id_split[0]
                        component_caused_by_id = component_id_split[1]
                    search_target_found = True
                    break

        trace.action_executions = action_executions
        trace.rules = rules
        trace.trigger_instances = trigger_instances
        return trace

    @staticmethod
    def _apply_display_filters(trace, args):
        """
        This function looks at the disaply filters to determine which components
        should be displayed.
        """
        # If all the filters are false nothing is to be filtered.
        all_component_types = not(args.show_executions or
                                  args.show_rules or
                                  args.show_trigger_instances)

        # check if noop_triggers are to be hidden. This check applies whenever TriggerInstances
        # are to be shown.
        if (all_component_types or args.show_trigger_instances) and args.hide_noop_triggers:
            filtered_trigger_instances = []
            for trigger_instance in trace.trigger_instances:
                is_noop_trigger_instance = True
                for rule in trace.rules:
                    caused_by_id = rule.get('caused_by', {}).get('id', None)
                    if caused_by_id == trigger_instance['object_id']:
                        is_noop_trigger_instance = False
                if not is_noop_trigger_instance:
                    filtered_trigger_instances.append(trigger_instance)
            trace.trigger_instances = filtered_trigger_instances

        if all_component_types:
            return trace

        if not args.show_executions:
            trace.action_executions = []

        if not args.show_rules:
            trace.rules = []

        if not args.show_trigger_instances:
            trace.trigger_instances = []

        return trace


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 __future__ import absolute_import
17
18			from st2client.models import Resource, Trace, TriggerInstance, Rule, LiveAction
19			from st2client.exceptions.operations import OperationFailureException
20			from st2client.formatters import table
21			from st2client.formatters import execution as execution_formatter
22			from st2client.commands import resource
23			from st2client.utils.date import format_isodate_for_user_timezone
24
25
26			TRACE_ATTRIBUTE_DISPLAY_ORDER = ['id', 'trace_tag', 'action_executions', 'rules',
27			'trigger_instances', 'start_timestamp']
28
29			TRACE_HEADER_DISPLAY_ORDER = ['id', 'trace_tag', 'start_timestamp']
30
31			TRACE_COMPONENT_DISPLAY_LABELS = ['id', 'type', 'ref', 'updated_at']
32
33			TRACE_DISPLAY_ATTRIBUTES = ['all']
34
35			TRIGGER_INSTANCE_DISPLAY_OPTIONS = [
36			'all',
37			'trigger-instances',
38			'trigger_instances',
39			'triggerinstances',
40			'triggers'
41			]
42
43			ACTION_EXECUTION_DISPLAY_OPTIONS = [
44			'all',
45			'executions',
46			'action-executions',
47			'action_executions',
48			'actionexecutions',
49			'actions'
50			]
51
52
53			class TraceBranch(resource.ResourceBranch):
54			def __init__(self, description, app, subparsers, parent_parser=None):
55			super(TraceBranch, self).__init__(
56			Trace, description, app, subparsers,
57			parent_parser=parent_parser,
58			read_only=True,
59			commands={
60			'list': TraceListCommand,
61			'get': TraceGetCommand
62			})
63
64
65			class SingleTraceDisplayMixin(object):
66
67			def print_trace_details(self, trace, args, **kwargs):
68			options = {'attributes': TRACE_ATTRIBUTE_DISPLAY_ORDER if args.json else
69			TRACE_HEADER_DISPLAY_ORDER}
70			options['json'] = args.json
71			options['yaml'] = args.yaml
72			options['attribute_transform_functions'] = self.attribute_transform_functions
73
74			formatter = execution_formatter.ExecutionResult
75
76			self.print_output(trace, formatter, **options)
77
78			# Everything should be printed if we are printing json.
79			if args.json or args.yaml:
80			return
81
82			components = []
83			if any(attr in args.attr for attr in TRIGGER_INSTANCE_DISPLAY_OPTIONS):
84			components.extend([Resource(**{'id': trigger_instance['object_id'],
85			'type': TriggerInstance._alias.lower(),
			0 ignored issues – show Coding Style Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report It seems like `_alias` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
86			'ref': trigger_instance['ref'],
87			'updated_at': trigger_instance['updated_at']})
88			for trigger_instance in trace.trigger_instances])
89			if any(attr in args.attr for attr in ['all', 'rules']):
90			components.extend([Resource(**{'id': rule['object_id'],
91			'type': Rule._alias.lower(),
			0 ignored issues – show Coding Style Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report It seems like `_alias` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
92			'ref': rule['ref'],
93			'updated_at': rule['updated_at']})
94			for rule in trace.rules])
95			if any(attr in args.attr for attr in ACTION_EXECUTION_DISPLAY_OPTIONS):
96			components.extend([Resource(**{'id': execution['object_id'],
97			'type': LiveAction._alias.lower(),
			0 ignored issues – show Coding Style Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report It seems like `_alias` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
98			'ref': execution['ref'],
99			'updated_at': execution['updated_at']})
100			for execution in trace.action_executions])
101			if components:
102			components.sort(key=lambda resource: resource.updated_at)
103			self.print_output(components, table.MultiColumnTable,
104			attributes=TRACE_COMPONENT_DISPLAY_LABELS,
105			json=args.json, yaml=args.yaml)
106
107
108			class TraceListCommand(resource.ResourceCommand, SingleTraceDisplayMixin):
109			display_attributes = ['id', 'uid', 'trace_tag', 'start_timestamp']
110
111			attribute_transform_functions = {
112			'start_timestamp': format_isodate_for_user_timezone
113			}
114
115			attribute_display_order = TRACE_ATTRIBUTE_DISPLAY_ORDER
116
117			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2017-04-18 09:25 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 `22`). 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...
118
119			self.default_limit = 50
120
121			super(TraceListCommand, self).__init__(
122			resource, 'list', 'Get the list of the %s most recent %s.' %
123			(self.default_limit, resource.get_plural_display_name().lower()),
124			args, *kwargs)
125
126			self.resource_name = resource.get_plural_display_name().lower()
127			self.group = self.parser.add_mutually_exclusive_group()
128			self.parser.add_argument('-n', '--last', type=int, dest='last',
129			default=self.default_limit,
130			help=('List N most recent %s. Use -n -1 to fetch the full result \
131			set.' % self.resource_name))
132			self.parser.add_argument('-s', '--sort', type=str, dest='sort_order',
133			default='descending',
134			help=('Sort %s by start timestamp, '
135			'asc\|ascending (earliest first) '
136			'or desc\|descending (latest first)' % self.resource_name))
137
138			# Filter options
139			self.group.add_argument('-c', '--trace-tag', help='Trace-tag to filter the list.')
140			self.group.add_argument('-e', '--execution', help='Execution to filter the list.')
141			self.group.add_argument('-r', '--rule', help='Rule to filter the list.')
142			self.group.add_argument('-g', '--trigger-instance',
143			help='TriggerInstance to filter the list.')
144			# Display options
145			self.parser.add_argument('-a', '--attr', nargs='+',
146			default=self.display_attributes,
147			help=('List of attributes to include in the '
148			'output. "all" will return all '
149			'attributes.'))
150			self.parser.add_argument('-w', '--width', nargs='+', type=int,
151			default=None,
152			help=('Set the width of columns in output.'))
153
154			@resource.add_auth_token_to_kwargs_from_cli
155			def run(self, args, **kwargs):
156			# Filtering options
157			if args.trace_tag:
158			kwargs['trace_tag'] = args.trace_tag
159			if args.trigger_instance:
160			kwargs['trigger_instance'] = args.trigger_instance
161			if args.execution:
162			kwargs['execution'] = args.execution
163			if args.rule:
164			kwargs['rule'] = args.rule
165
166			if args.sort_order:
167			if args.sort_order in ['asc', 'ascending']:
168			kwargs['sort_asc'] = True
169			elif args.sort_order in ['desc', 'descending']:
170			kwargs['sort_desc'] = True
171			return self.manager.query_with_count(limit=args.last, **kwargs)
172
173			def run_and_print(self, args, **kwargs):
174			instances, count = self.run(args, **kwargs)
175
176			if instances and len(instances) == 1:
177			# For a single Trace we must include the components unless
178			# user has overriden the attributes to display
179			if args.attr == self.display_attributes:
180			args.attr = ['all']
181			self.print_trace_details(trace=instances[0], args=args)
182
183			if not args.json and not args.yaml:
184			if args.last and count and count > args.last:
185			table.SingleRowTable.note_box(self.resource_name, 1)
			0 ignored issues – show Coding Style introduced 2017-07-12 02:34 UTC by Report Bug Copy Issue Report The indentation here looks off. 20 spaces were expected, but 24 were found. Loading history...
186			else:
187			if args.json or args.yaml:
188			self.print_output(instances, table.MultiColumnTable,
189			attributes=args.attr, widths=args.width,
190			json=args.json, yaml=args.yaml,
191			attribute_transform_functions=self.attribute_transform_functions)
192			else:
193			self.print_output(instances, table.MultiColumnTable,
194			attributes=args.attr, widths=args.width,
195			attribute_transform_functions=self.attribute_transform_functions)
196
197			if args.last and count and count > args.last:
198			table.SingleRowTable.note_box(self.resource_name, args.last)
199
200
201			class TraceGetCommand(resource.ResourceGetCommand, SingleTraceDisplayMixin):
202			display_attributes = ['all']
203			attribute_display_order = TRACE_ATTRIBUTE_DISPLAY_ORDER
204			attribute_transform_functions = {
205			'start_timestamp': format_isodate_for_user_timezone
206			}
207
208			pk_argument_name = 'id'
209
210			def __init__(self, resource, args, *kwargs):
			0 ignored issues – show Comprehensibility Bug introduced 2017-04-18 09:25 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 `22`). 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...
211			super(TraceGetCommand, self).__init__(resource, args, *kwargs)
212
213			# Causation chains
214			self.causation_group = self.parser.add_mutually_exclusive_group()
215
216			self.causation_group.add_argument('-e', '--execution',
217			help='Execution to show causation chain.')
218			self.causation_group.add_argument('-r', '--rule', help='Rule to show causation chain.')
219			self.causation_group.add_argument('-g', '--trigger-instance',
220			help='TriggerInstance to show causation chain.')
221
222			# display filter group
223			self.display_filter_group = self.parser.add_argument_group()
224
225			self.display_filter_group.add_argument('--show-executions', action='store_true',
226			help='Only show executions.')
227			self.display_filter_group.add_argument('--show-rules', action='store_true',
228			help='Only show rules.')
229			self.display_filter_group.add_argument('--show-trigger-instances', action='store_true',
230			help='Only show trigger instances.')
231			self.display_filter_group.add_argument('-n', '--hide-noop-triggers', action='store_true',
232			help='Hide noop trigger instances.')
233
234			@resource.add_auth_token_to_kwargs_from_cli
235			def run(self, args, **kwargs):
236			resource_id = getattr(args, self.pk_argument_name, None)
237			return self.get_resource_by_id(resource_id, **kwargs)
238
239			@resource.add_auth_token_to_kwargs_from_cli
240			def run_and_print(self, args, **kwargs):
241			trace = None
242			try:
243			trace = self.run(args, **kwargs)
244			except resource.ResourceNotFoundError:
245			self.print_not_found(args.id)
246			raise OperationFailureException('Trace %s not found.' % (args.id))
247			# First filter for causation chains
248			trace = self._filter_trace_components(trace=trace, args=args)
249			# next filter for display purposes
250			trace = self._apply_display_filters(trace=trace, args=args)
251			return self.print_trace_details(trace=trace, args=args)
252
253			@staticmethod
254			def _filter_trace_components(trace, args):
255			"""
256			This function walks up the component causal chain. It only returns
257			properties in the causal chain and nothing else.
258			"""
259			# check if any filtering is desired
260			if not (args.execution or args.rule or args.trigger_instance):
261			return trace
262
263			component_id = None
264			component_type = None
265
266			# pick the right component type
267			if args.execution:
268			component_id = args.execution
269			component_type = 'action_execution'
270			elif args.rule:
271			component_id = args.rule
272			component_type = 'rule'
273			elif args.trigger_instance:
274			component_id = args.trigger_instance
275			component_type = 'trigger_instance'
276
277			# Initialize collection to use
278			action_executions = []
279			rules = []
280			trigger_instances = []
281
282			# setup flag to properly manage termination conditions
283			search_target_found = component_id and component_type
284
285			while search_target_found:
286			components_list = []
287			if component_type == 'action_execution':
288			components_list = trace.action_executions
289			to_update_list = action_executions
290			elif component_type == 'rule':
291			components_list = trace.rules
292			to_update_list = rules
293			elif component_type == 'trigger_instance':
294			components_list = trace.trigger_instances
295			to_update_list = trigger_instances
296			# Look for search_target in the right collection and
297			# once found look up the caused_by to keep movig up
298			# the chain.
299			search_target_found = False
300			# init to default value
301			component_caused_by_id = None
302			for component in components_list:
303			test_id = component['object_id']
304			if test_id == component_id:
305			caused_by = component.get('caused_by', {})
306			component_id = caused_by.get('id', None)
307			component_type = caused_by.get('type', None)
308			# If provided the component_caused_by_id must match as well. This is mostly
309			# applicable for rules since the same rule may appear multiple times and can
310			# only be distinguished by causing TriggerInstance.
311			if component_caused_by_id and component_caused_by_id != component_id:
312			continue
313			component_caused_by_id = None
314			to_update_list.append(component)
315			# In some cases the component_id and the causing component are combined to
316			# provide the complete causation chain. Think rule + triggerinstance
317			if component_id and ':' in component_id:
318			component_id_split = component_id.split(':')
319			component_id = component_id_split[0]
320			component_caused_by_id = component_id_split[1]
321			search_target_found = True
322			break
323
324			trace.action_executions = action_executions
325			trace.rules = rules
326			trace.trigger_instances = trigger_instances
327			return trace
328
329			@staticmethod
330			def _apply_display_filters(trace, args):
331			"""
332			This function looks at the disaply filters to determine which components
333			should be displayed.
334			"""
335			# If all the filters are false nothing is to be filtered.
336			all_component_types = not(args.show_executions or
337			args.show_rules or
338			args.show_trigger_instances)
339
340			# check if noop_triggers are to be hidden. This check applies whenever TriggerInstances
341			# are to be shown.
342			if (all_component_types or args.show_trigger_instances) and args.hide_noop_triggers:
343			filtered_trigger_instances = []
344			for trigger_instance in trace.trigger_instances:
345			is_noop_trigger_instance = True
346			for rule in trace.rules:
347			caused_by_id = rule.get('caused_by', {}).get('id', None)
348			if caused_by_id == trigger_instance['object_id']:
349			is_noop_trigger_instance = False
350			if not is_noop_trigger_instance:
351			filtered_trigger_instances.append(trigger_instance)
352			trace.trigger_instances = filtered_trigger_instances
353
354			if all_component_types:
355			return trace
356
357			if not args.show_executions:
358			trace.action_executions = []
359
360			if not args.show_rules:
361			trace.rules = []
362
363			if not args.show_trigger_instances:
364			trace.trigger_instances = []
365
366			return trace
367

StackStorm / st2

Pull Request — master (#4194)

TraceListCommand.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TraceListCommand.init() A