st2common.services.get_parent_context() - Code Metrics - Inspection of "Merge branch 'master' of https://github.com/StackS..." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — kale/action-datastore ( 9f47ff...fd1ef0 )

unknown

created 2016-02-02 17:08 UTC

st2common.services.get_parent_context() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	11
rs	9.4285
cc	2

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

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

from st2common import log as logging
from st2common.util import reference
import st2common.util.action_db as action_utils
from st2common.constants import action as action_constants
from st2common.persistence.execution import ActionExecution
from st2common.persistence.runner import RunnerType
from st2common.persistence.rule import Rule
from st2common.persistence.trigger import TriggerType, Trigger, TriggerInstance
from st2common.models.api.action import RunnerTypeAPI, ActionAPI, LiveActionAPI
from st2common.models.api.rule import RuleAPI
from st2common.models.api.trigger import TriggerTypeAPI, TriggerAPI, TriggerInstanceAPI
from st2common.models.db.execution import ActionExecutionDB

__all__ = [
    'create_execution_object',
    'update_execution',
    'abandon_execution_if_incomplete',
    'is_execution_canceled',
    'AscendingSortedDescendantView',
    'DFSDescendantView',
    'get_descendants'
]

LOG = logging.getLogger(__name__)

SKIPPED = ['id', 'callback', 'action', 'runner_info', 'parameters']


def _decompose_liveaction(liveaction_db):
    """
    Splits the liveaction into an ActionExecution compatible dict.
    """
    decomposed = {'liveaction': {}}
    liveaction_api = vars(LiveActionAPI.from_model(liveaction_db))
    for k in liveaction_api.keys():
        if k in SKIPPED:
            decomposed['liveaction'][k] = liveaction_api[k]
        else:
            decomposed[k] = getattr(liveaction_db, k)
    return decomposed


def create_execution_object(liveaction, publish=True):
    action_db = action_utils.get_action_by_ref(liveaction.action)
    runner = RunnerType.get_by_name(action_db.runner_type['name'])

    attrs = {
        'action': vars(ActionAPI.from_model(action_db)),
        'parameters': liveaction['parameters'],
        'runner': vars(RunnerTypeAPI.from_model(runner))
    }
    attrs.update(_decompose_liveaction(liveaction))

    if 'rule' in liveaction.context:
        rule = reference.get_model_from_ref(Rule, liveaction.context.get('rule', {}))
        attrs['rule'] = vars(RuleAPI.from_model(rule))

    if 'trigger_instance' in liveaction.context:
        trigger_instance_id = liveaction.context.get('trigger_instance', {})
        trigger_instance_id = trigger_instance_id.get('id', None)
        trigger_instance = TriggerInstance.get_by_id(trigger_instance_id)
        trigger = reference.get_model_by_resource_ref(db_api=Trigger,
                                                      ref=trigger_instance.trigger)
        trigger_type = reference.get_model_by_resource_ref(db_api=TriggerType,
                                                           ref=trigger.type)
        trigger_instance = reference.get_model_from_ref(
            TriggerInstance, liveaction.context.get('trigger_instance', {}))
        attrs['trigger_instance'] = vars(TriggerInstanceAPI.from_model(trigger_instance))
        attrs['trigger'] = vars(TriggerAPI.from_model(trigger))
        attrs['trigger_type'] = vars(TriggerTypeAPI.from_model(trigger_type))

    parent = _get_parent_execution(liveaction)
    if parent:
        attrs['parent'] = str(parent.id)

    execution = ActionExecutionDB(**attrs)
    execution = ActionExecution.add_or_update(execution, publish=publish)

    if parent:
        if str(execution.id) not in parent.children:
            parent.children.append(str(execution.id))
            ActionExecution.add_or_update(parent)

    return execution


def _get_parent_execution(child_liveaction_db):
    parent_context = child_liveaction_db.context.get('parent', None)

    if parent_context:
        parent_id = parent_context['execution_id']
        try:
            return ActionExecution.get_by_id(parent_id)
        except:
            LOG.exception('No valid execution object found in db for id: %s' % parent_id)

            return None
    return None


def update_execution(liveaction_db, publish=True):
    execution = ActionExecution.get(liveaction__id=str(liveaction_db.id))
    decomposed = _decompose_liveaction(liveaction_db)
    for k, v in six.iteritems(decomposed):
        setattr(execution, k, v)
    execution = ActionExecution.add_or_update(execution, publish=publish)
    return execution


def abandon_execution_if_incomplete(liveaction_id, publish=True):
    """
    Marks execution as abandoned if it is still incomplete. Abandoning an
    execution implies that its end state is unknown and cannot anylonger
    be determined. This method should only be called if the owning process
    is certain it can no longer determine status of an execution.
    """
    liveaction_db = action_utils.get_liveaction_by_id(liveaction_id)
    # No need to abandon and already complete action
    if liveaction_db.status in action_constants.LIVEACTION_COMPLETED_STATES:
        raise ValueError('LiveAction %s already in a completed state %s.' %
                         (liveaction_id, liveaction_db.status))
    liveaction_db = action_utils.update_liveaction_status(
        status=action_constants.LIVEACTION_STATUS_ABANDONED,
        liveaction_db=liveaction_db,
        result={})
    execution_db = update_execution(liveaction_db, publish=publish)
    LOG.info('Marked execution %s as %s.', execution_db.id,
             action_constants.LIVEACTION_STATUS_ABANDONED)
    return execution_db


def is_execution_canceled(execution_id):
    try:
        execution = ActionExecution.get_by_id(execution_id)
        return execution.status == action_constants.LIVEACTION_STATUS_CANCELED
    except:
        return False  # XXX: What to do here?


def get_parent_context(liveaction_db):
    """
    Returns context of the parent execution.

    :return: If found the parent context else None.
    :rtype: dict
    """
    context = getattr(liveaction_db, 'context', None)
    if not context:
        return None
    return context.get('parent', None)


class AscendingSortedDescendantView(object):
    def __init__(self):
        self._result = []

    def add(self, child):
        self._result.append(child)

    @property
    def result(self):
        return sorted(self._result, key=lambda execution: execution.start_timestamp)


class DFSDescendantView(object):
    def __init__(self):
        self._result = []

    def add(self, child):
        self._result.append(child)

    @property
    def result(self):
        return self._result


DESCENDANT_VIEWS = {
    'sorted': AscendingSortedDescendantView,
    'default': DFSDescendantView
}


def get_descendants(actionexecution_id, descendant_depth=-1, result_fmt=None):
    """
    Returns all descendant executions upto the specified descendant_depth for
    the supplied actionexecution_id.
    """
    descendants = DESCENDANT_VIEWS.get(result_fmt, DFSDescendantView)()
    children = ActionExecution.query(parent=actionexecution_id,
                                     **{'order_by': ['start_timestamp']})
    LOG.debug('Found %s children for id %s.', len(children), actionexecution_id)
    current_level = [(child, 1) for child in children]

    while current_level:
        parent, level = current_level.pop(0)
        parent_id = str(parent.id)
        descendants.add(parent)
        if not parent.children:
            continue
        if level != -1 and level == descendant_depth:
            continue
        children = ActionExecution.query(parent=parent_id, **{'order_by': ['start_timestamp']})
        LOG.debug('Found %s children for id %s.', len(children), parent_id)
        # prepend for DFS
        for idx in range(len(children)):
            current_level.insert(idx, (children[idx], level + 1))
    return descendants.result


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			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
17			# contributor license agreements. See the NOTICE file distributed with
18			# this work for additional information regarding copyright ownership.
19			# The ASF licenses this file to You under the Apache License, Version 2.0
20			# (the "License"); you may not use this file except in compliance with
21			# the License. You may obtain a copy of the License at
22			#
23			# http://www.apache.org/licenses/LICENSE-2.0
24			#
25			# Unless required by applicable law or agreed to in writing, software
26			# distributed under the License is distributed on an "AS IS" BASIS,
27			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
28			# See the License for the specific language governing permissions and
29			# limitations under the License.
30
31			import six
32
33			from st2common import log as logging
34			from st2common.util import reference
35			import st2common.util.action_db as action_utils
36			from st2common.constants import action as action_constants
37			from st2common.persistence.execution import ActionExecution
38			from st2common.persistence.runner import RunnerType
39			from st2common.persistence.rule import Rule
40			from st2common.persistence.trigger import TriggerType, Trigger, TriggerInstance
41			from st2common.models.api.action import RunnerTypeAPI, ActionAPI, LiveActionAPI
42			from st2common.models.api.rule import RuleAPI
43			from st2common.models.api.trigger import TriggerTypeAPI, TriggerAPI, TriggerInstanceAPI
44			from st2common.models.db.execution import ActionExecutionDB
45
46			__all__ = [
47			'create_execution_object',
48			'update_execution',
49			'abandon_execution_if_incomplete',
50			'is_execution_canceled',
51			'AscendingSortedDescendantView',
52			'DFSDescendantView',
53			'get_descendants'
54			]
55
56			LOG = logging.getLogger(__name__)
57
58			SKIPPED = ['id', 'callback', 'action', 'runner_info', 'parameters']
59
60
61			def _decompose_liveaction(liveaction_db):
62			"""
63			Splits the liveaction into an ActionExecution compatible dict.
64			"""
65			decomposed = {'liveaction': {}}
66			liveaction_api = vars(LiveActionAPI.from_model(liveaction_db))
67			for k in liveaction_api.keys():
68			if k in SKIPPED:
69			decomposed['liveaction'][k] = liveaction_api[k]
70			else:
71			decomposed[k] = getattr(liveaction_db, k)
72			return decomposed
73
74
75			def create_execution_object(liveaction, publish=True):
76			action_db = action_utils.get_action_by_ref(liveaction.action)
77			runner = RunnerType.get_by_name(action_db.runner_type['name'])
78
79			attrs = {
80			'action': vars(ActionAPI.from_model(action_db)),
81			'parameters': liveaction['parameters'],
82			'runner': vars(RunnerTypeAPI.from_model(runner))
83			}
84			attrs.update(_decompose_liveaction(liveaction))
85
86			if 'rule' in liveaction.context:
87			rule = reference.get_model_from_ref(Rule, liveaction.context.get('rule', {}))
88			attrs['rule'] = vars(RuleAPI.from_model(rule))
89
90			if 'trigger_instance' in liveaction.context:
91			trigger_instance_id = liveaction.context.get('trigger_instance', {})
92			trigger_instance_id = trigger_instance_id.get('id', None)
93			trigger_instance = TriggerInstance.get_by_id(trigger_instance_id)
94			trigger = reference.get_model_by_resource_ref(db_api=Trigger,
95			ref=trigger_instance.trigger)
96			trigger_type = reference.get_model_by_resource_ref(db_api=TriggerType,
97			ref=trigger.type)
98			trigger_instance = reference.get_model_from_ref(
99			TriggerInstance, liveaction.context.get('trigger_instance', {}))
100			attrs['trigger_instance'] = vars(TriggerInstanceAPI.from_model(trigger_instance))
101			attrs['trigger'] = vars(TriggerAPI.from_model(trigger))
102			attrs['trigger_type'] = vars(TriggerTypeAPI.from_model(trigger_type))
103
104			parent = _get_parent_execution(liveaction)
105			if parent:
106			attrs['parent'] = str(parent.id)
107
108			execution = ActionExecutionDB(**attrs)
109			execution = ActionExecution.add_or_update(execution, publish=publish)
110
111			if parent:
112			if str(execution.id) not in parent.children:
113			parent.children.append(str(execution.id))
114			ActionExecution.add_or_update(parent)
115
116			return execution
117
118
119			def _get_parent_execution(child_liveaction_db):
120			parent_context = child_liveaction_db.context.get('parent', None)
121
122			if parent_context:
123			parent_id = parent_context['execution_id']
124			try:
125			return ActionExecution.get_by_id(parent_id)
126			except:
127			LOG.exception('No valid execution object found in db for id: %s' % parent_id)
			0 ignored issues – show Coding Style Best Practice introduced 2016-01-14 18:36 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
128			return None
129			return None
130
131
132			def update_execution(liveaction_db, publish=True):
133			execution = ActionExecution.get(liveaction__id=str(liveaction_db.id))
134			decomposed = _decompose_liveaction(liveaction_db)
135			for k, v in six.iteritems(decomposed):
136			setattr(execution, k, v)
137			execution = ActionExecution.add_or_update(execution, publish=publish)
138			return execution
139
140
141			def abandon_execution_if_incomplete(liveaction_id, publish=True):
142			"""
143			Marks execution as abandoned if it is still incomplete. Abandoning an
144			execution implies that its end state is unknown and cannot anylonger
145			be determined. This method should only be called if the owning process
146			is certain it can no longer determine status of an execution.
147			"""
148			liveaction_db = action_utils.get_liveaction_by_id(liveaction_id)
149			# No need to abandon and already complete action
150			if liveaction_db.status in action_constants.LIVEACTION_COMPLETED_STATES:
151			raise ValueError('LiveAction %s already in a completed state %s.' %
152			(liveaction_id, liveaction_db.status))
153			liveaction_db = action_utils.update_liveaction_status(
154			status=action_constants.LIVEACTION_STATUS_ABANDONED,
155			liveaction_db=liveaction_db,
156			result={})
157			execution_db = update_execution(liveaction_db, publish=publish)
158			LOG.info('Marked execution %s as %s.', execution_db.id,
159			action_constants.LIVEACTION_STATUS_ABANDONED)
160			return execution_db
161
162
163			def is_execution_canceled(execution_id):
164			try:
165			execution = ActionExecution.get_by_id(execution_id)
166			return execution.status == action_constants.LIVEACTION_STATUS_CANCELED
167			except:
168			return False # XXX: What to do here?
169
170
171			def get_parent_context(liveaction_db):
172			"""
173			Returns context of the parent execution.
174
175			:return: If found the parent context else None.
176			:rtype: dict
177			"""
178			context = getattr(liveaction_db, 'context', None)
179			if not context:
180			return None
181			return context.get('parent', None)
182
183
184			class AscendingSortedDescendantView(object):
185			def __init__(self):
186			self._result = []
187
188			def add(self, child):
189			self._result.append(child)
190
191			@property
192			def result(self):
193			return sorted(self._result, key=lambda execution: execution.start_timestamp)
194
195
196			class DFSDescendantView(object):
197			def __init__(self):
198			self._result = []
199
200			def add(self, child):
201			self._result.append(child)
202
203			@property
204			def result(self):
205			return self._result
206
207
208			DESCENDANT_VIEWS = {
209			'sorted': AscendingSortedDescendantView,
210			'default': DFSDescendantView
211			}
212
213
214			def get_descendants(actionexecution_id, descendant_depth=-1, result_fmt=None):
215			"""
216			Returns all descendant executions upto the specified descendant_depth for
217			the supplied actionexecution_id.
218			"""
219			descendants = DESCENDANT_VIEWS.get(result_fmt, DFSDescendantView)()
220			children = ActionExecution.query(parent=actionexecution_id,
221			**{'order_by': ['start_timestamp']})
222			LOG.debug('Found %s children for id %s.', len(children), actionexecution_id)
223			current_level = [(child, 1) for child in children]
224
225			while current_level:
226			parent, level = current_level.pop(0)
227			parent_id = str(parent.id)
228			descendants.add(parent)
229			if not parent.children:
230			continue
231			if level != -1 and level == descendant_depth:
232			continue
233			children = ActionExecution.query(parent=parent_id, **{'order_by': ['start_timestamp']})
234			LOG.debug('Found %s children for id %s.', len(children), parent_id)
235			# prepend for DFS
236			for idx in range(len(children)):
237			current_level.insert(idx, (children[idx], level + 1))
238			return descendants.result
239

Plexxi / st2

GitHub Access Token became invalid

Push — kale/action-datastore ( 9f47ff...fd1ef0 )

st2common.services.get_parent_context() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like