WorkflowDispatcher.process() - Code Metrics - Inspection of "Implement the Orchestra runner" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Setup Failed

Pull Request — master (#4154)

by W

created 2018-06-05 04:04 UTC

WorkflowDispatcher.process() B

↳ Parent: WorkflowDispatcher

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	6
dl	0
loc	32
rs	7.5384
c	1
b	0
f	0

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

from __future__ import absolute_import

import kombu

from orchestra import states

from st2common import log as logging
from st2common.models.db import workflow as wf_db_models
from st2common.services import workflows as wf_svc
from st2common.transport import consumers
from st2common.transport import queues
from st2common.transport import utils as txpt_utils


LOG = logging.getLogger(__name__)


WORKFLOW_EXECUTION_QUEUES = [
    queues.WORKFLOW_EXECUTION_WORK_QUEUE,
    queues.WORKFLOW_EXECUTION_RESUME_QUEUE
]


class WorkflowDispatcher(consumers.MessageHandler):
    message_type = wf_db_models.WorkflowExecutionDB

    def __init__(self, connection, queues):
param = 5

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

    def get_queue_consumer(self, connection, queues):
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        # We want to use a special ActionsQueueConsumer which uses 2 dispatcher pools
        return consumers.QueueConsumer(connection=connection, queues=queues, handler=self)

    def process(self, wf_ex_db):
        # Refresh record from the database in case the request is in the queue for too long.
        conductor, wf_ex_db = wf_svc.refresh_conductor(str(wf_ex_db.id))

        # Continue if workflow is still active.
        if conductor.get_workflow_state() not in states.COMPLETED_STATES:
            # Set workflow to running state.
            conductor.set_workflow_state(states.RUNNING)

            # Identify the next set of tasks to execute.
            next_tasks = conductor.get_next_tasks()

            # Mark the tasks as running in the task flow before actual task execution.
            for task in next_tasks:
                conductor.update_task_flow(task['id'], states.RUNNING)

        # Update workflow execution and related liveaction and action execution.
        wf_svc.update_execution_records(wf_ex_db, conductor)

        # If workflow execution is no longer active, then stop processing here.
        if wf_ex_db.status in states.COMPLETED_STATES:
            return

        # Request task execution for the tasks.
        for task in next_tasks:
            try:
                task_id, task_spec, task_ctx = task['id'], task['spec'], task['ctx']
                st2_ctx = {'execution_id': wf_ex_db.action_execution}
                wf_svc.request_task_execution(wf_ex_db, task_id, task_spec, task_ctx, st2_ctx)
            except Exception as e:
                wf_svc.fail_workflow_execution(str(wf_ex_db.id), e, task_id=task['id'])
                break


def get_engine():
    with kombu.Connection(txpt_utils.get_messaging_urls()) as conn:
        return WorkflowDispatcher(conn, WORKFLOW_EXECUTION_QUEUES)


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			import kombu
19
20			from orchestra import states
21
22			from st2common import log as logging
23			from st2common.models.db import workflow as wf_db_models
24			from st2common.services import workflows as wf_svc
25			from st2common.transport import consumers
26			from st2common.transport import queues
27			from st2common.transport import utils as txpt_utils
28
29
30			LOG = logging.getLogger(__name__)
31
32
33			WORKFLOW_EXECUTION_QUEUES = [
34			queues.WORKFLOW_EXECUTION_WORK_QUEUE,
35			queues.WORKFLOW_EXECUTION_RESUME_QUEUE
36			]
37
38
39			class WorkflowDispatcher(consumers.MessageHandler):
40			message_type = wf_db_models.WorkflowExecutionDB
41
42			def __init__(self, connection, queues):
			0 ignored issues – show Comprehensibility Bug introduced 2018-05-31 05:58 UTC by Report Bug Copy Issue Report `queues` is re-defining a name which is already available in the outer-scope (previously defined on line `26`). 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...
43			super(WorkflowDispatcher, self).__init__(connection, queues)
44
45			def get_queue_consumer(self, connection, queues):
			0 ignored issues – show Comprehensibility Bug introduced 2018-05-31 05:58 UTC by Report Bug Copy Issue Report `queues` is re-defining a name which is already available in the outer-scope (previously defined on line `26`). 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...
46			# We want to use a special ActionsQueueConsumer which uses 2 dispatcher pools
47			return consumers.QueueConsumer(connection=connection, queues=queues, handler=self)
48
49			def process(self, wf_ex_db):
50			# Refresh record from the database in case the request is in the queue for too long.
51			conductor, wf_ex_db = wf_svc.refresh_conductor(str(wf_ex_db.id))
52
53			# Continue if workflow is still active.
54			if conductor.get_workflow_state() not in states.COMPLETED_STATES:
55			# Set workflow to running state.
56			conductor.set_workflow_state(states.RUNNING)
57
58			# Identify the next set of tasks to execute.
59			next_tasks = conductor.get_next_tasks()
60
61			# Mark the tasks as running in the task flow before actual task execution.
62			for task in next_tasks:
63			conductor.update_task_flow(task['id'], states.RUNNING)
64
65			# Update workflow execution and related liveaction and action execution.
66			wf_svc.update_execution_records(wf_ex_db, conductor)
67
68			# If workflow execution is no longer active, then stop processing here.
69			if wf_ex_db.status in states.COMPLETED_STATES:
70			return
71
72			# Request task execution for the tasks.
73			for task in next_tasks:
74			try:
75			task_id, task_spec, task_ctx = task['id'], task['spec'], task['ctx']
76			st2_ctx = {'execution_id': wf_ex_db.action_execution}
77			wf_svc.request_task_execution(wf_ex_db, task_id, task_spec, task_ctx, st2_ctx)
78			except Exception as e:
79			wf_svc.fail_workflow_execution(str(wf_ex_db.id), e, task_id=task['id'])
80			break
81
82
83			def get_engine():
84			with kombu.Connection(txpt_utils.get_messaging_urls()) as conn:
85			return WorkflowDispatcher(conn, WORKFLOW_EXECUTION_QUEUES)
86

StackStorm / st2

Pull Request — master (#4154)

WorkflowDispatcher.process() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like