TraceListCommand.__init__() - Code Metrics - Inspection of "Experimental/circle 2.0 (with workflows)" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#3561)

by Arma

created 2017-07-13 18:35 UTC

TraceListCommand.init() B

↳ Parent: TraceListCommand

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
dl	0
loc	33
rs	8.8571
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 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
        super(TraceListCommand, self).__init__(
            resource, 'list', 'Get the list of the 50 most recent %s.' %
            resource.get_plural_display_name().lower(),
            *args, **kwargs)

        self.default_limit = 50
        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.' % 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
        result, count = self.manager.query(limit=args.last, **kwargs)
        return (result, count)

    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 int(count) > args.last:
                        table.SingleRowTable.note_box(self.resource_name, 1)

        else:
            if args.json or args.yaml:
                self.print_output(reversed(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(reversed(instances), table.MultiColumnTable,
                                  attributes=args.attr, widths=args.width,
                                  attribute_transform_functions=self.attribute_transform_functions)

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

StackStorm / st2

Pull Request — master (#3561)

TraceListCommand.__init__() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TraceListCommand.init() B