StepState.check_requirements() - Code Metrics - sportarchive/CloudProcessingEngine-Decider - Measure and Improve Code Quality continuously with Scrutinizer

StepState.check_requirements() F
last analyzed 2016-11-10 17:43 UTC

↳ Parent: StepState

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	10
dl	0
loc	35
rs	3.1304
c	1
b	0
f	0

How to fix Complexity

from __future__ import (
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    absolute_import,
    division,
    print_function
)

import weakref
import logging
import collections

from .state_status import (
    StateStatus,
    StepStateStatus,
)
from .step import Step

_LOGGER = logging.getLogger(__name__)
INIT_STEP = '$init'
END_STEP = '$end'


class DeciderStepResult(object):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    pass


class DeciderStep(Step):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    def run(self, _step_input):
        return DeciderStepResult()

    def prepare(self, _context):
        return None

    def render(self, output):
        # The out of the INIT step is the input to the workflow
        return output


class State(object):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    __slots__ = (
        'status',
        'step_states',
        '_context',
        '_orphaned_steps',
        '_init_step',
        '_end_step',
    )

    def __init__(self):
        """Init state"""
        self.status = StateStatus.init
        self.step_states = {}
        self._context = None
        self._orphaned_steps = {}
        # Add an INIT/END fake steps
        self._init_step = StepState(
            step=DeciderStep(INIT_STEP),
            status='running',
            context='__init__'
        )
        self._end_step = StepState(
            step=DeciderStep(END_STEP,
                             requires=[(INIT_STEP, 'completed')]),
            context='__init__'
        )
        self._stepstate_insert(self._init_step)
        self._stepstate_insert(self._end_step)

    def __repr__(self):
        return '{ctype}(status={status},steps={steps})'.format(
            ctype=self.__class__.__name__,
            status=self.status.name,
            steps=len(self.step_states)
        )

    #################################################
    def is_in_state(self, state_name):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        desired_state = getattr(StateStatus, state_name)
        return self.status.means(desired_state)

    #################################################
    # Context management
    def __call__(self, context):
        self._context = context
        return self

    def __enter__(self):
        assert self._context is not None

    def __exit__(self, exc_type, exc_value, traceback):
        assert self._context is not None
        # Only clear the context is we didn't encounter an exception. Otherwise
        # keep it for debug purposes
        if exc_type is None:
            self._context = None

    #################################################
    # State manipulations
    def set_abort(self):
        """Abort the state machine.
        """
        assert self._context is not None
        self.step_update(END_STEP, 'aborted')

    def set_input(self, input_data):
        """Set the input of the state machine.
        """
        assert self._context is not None
        assert self.status is StateStatus.init

        self.step_update(INIT_STEP, 'completed', new_data=input_data)
        self.status = StateStatus.running

    def step_update(self, step_name, new_status, new_data=None):
        """Update a Step with new status and, optionally, output data.
        """
        assert self._context is not None
        step_state = self.step_states[step_name]
        step_state.update(new_status,
                          context=self._context,
                          new_output=new_data)

        if self._end_step.status is StepStateStatus.ready:
            self.status = StateStatus.succeeded
        elif self._end_step.status is StepStateStatus.aborted:
            self.status = StateStatus.failed

    def step_insert(self, step):
        """Add a step definition to the state.
        """
        assert self._context is not None
        step_state = StepState(step, self._context)
        _LOGGER.info('Defining new step %r in state', step_state)

        # All steps are children of the root INIT_STEP step
        self._init_step.children.add(step_state)
        step_state.parents.add(self._init_step)

        if self._stepstate_insert(step_state):
            # All steps are a parent of END_STEP
            # FIXME: Could be optimized so that only steps without children are

            #        parented to END_STEP
            self._end_step.parents.add(step_state)
            self._end_step.step.requires[step_state.step.name] = \
                StepStateStatus.completed

            step_state.children.add(self._end_step)
            # See if we can re-parents some previously orphaned Step with the
            # one we have just added.
            orphans = self._orphaned_steps.pop(step.name, [])
            for orphan in orphans:
                self._stepstate_insert(orphan)

    #################################################
    def step_next(self, hint=None):
        """Search for steps ready to be ran.

        :param str hint:
            Place to start searching steps that are now ready to be scheduled
            (usually, the last `completed` step).
        """
        ready_steps = set()
        # Were we given a place to start looking?
        if hint:
            hint_step = self.step_states[hint]

            # Check that this is completed
            if not hint_step.is_completed:
                # FIXME: This should be an error

                return set()

            for child in hint_step.children:
                if child.status is StepStateStatus.ready:
                    ready_steps.add(child)

        else:
            # Walk the whole tree collecting ready StepState
            for child in self._init_step.children:
                if child.status is StepStateStatus.ready:
                    ready_steps.add(child)
                else:
                    child_name = child.step.name
                    ready_steps |= self.step_next(child_name)

        return ready_steps

    def _stepstate_insert(self, step_state):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        if not step_state.step.requires:
            self.step_states[step_state.name] = step_state
            return True

        elif all([required in self.step_states
                  for required in step_state.step.requires.keys()]):
            # All the required steps are defined, update their children set and
            # record them in this step_state's parents set
            for required in step_state.step.requires.keys():
                self.step_states[required].children.add(step_state)
                step_state.parents.add(self.step_states[required])

            self.step_states[step_state.name] = step_state
            return True

        else:
            # We failed to insert this step_state
            # Parent steps are missing, set it as orphaned
            for required in step_state.step.requires.keys():
                if required not in self.step_states:
                    self._orphaned_steps.setdefault(
                        required, set()
                    ).add(step_state)
            return False


class StepState(object):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""

    __slots__ = ('status',
                 'step',
                 'input',
                 'output',
                 'children',
                 'parents',
                 'history',
                 '__weakref__')

    def __init__(self, step, context,
                 status=StepStateStatus.pending):

        if not isinstance(status, StepStateStatus):
            status = getattr(StepStateStatus, status)

        self.step = step
        self.status = status
        self.input = None
        self.output = None
        self.children = weakref.WeakSet()
        self.parents = weakref.WeakSet()
        self.history = collections.deque()
        self.history.append(
            (status, context)
        )

    def __repr__(self):
        return 'StepState({name}:{status})'.format(
            name=self.step.name, status=self.status.name
        )

    @property
    def name(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        return self.step.name

    @property
    def is_completed(self):
        """`True` if the status is either `succeeded`, `failed` or `skipped`.
        """
        return self.status.means(StepStateStatus.completed)

    def _prepare(self):
        """Prepare the input from the step input template and the parents data.
        """
        assert self.status is StepStateStatus.ready
        context = self._build_context()
        render = self.step.prepare(context)
        return render

    def _build_context(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        context = {}
        for parent in self.parents:
            if parent.name is INIT_STEP:
                context.update({'__input__': parent.output})
            else:
                context.update({parent.name: parent.output})
        return context

    def _record(self, output):
        """Invoke the step rendering of the results."""
        attrs = self.step.render(output)
        self.output = attrs

    def check_requirements(self, context):
        """`True` if the Step is ready to be evaluated.

        This means the step hasn't ran yet and all its parents are completed.
        """
        if self.status is StepStateStatus.ready:
            return
        # You cannot be ready if you are already completed/aborted
        if self.status is not StepStateStatus.pending:
            return

        # Check that all our parents are completed per the step's requirements.
        _LOGGER.info('Step %r requirements %r', self, self.step.requires)

        ready = True
        for parent in self.parents:
            if parent.name is INIT_STEP:
                continue
            assert parent.name in self.step.requires

            if not parent.is_completed:
                ready = False
                continue

            req_status = self.step.requires[parent.name]
            _LOGGER.info('Checking parent %r meets requirement %r',
                          parent, req_status)

            if not parent.status.means(req_status):
                self.update('skipped', context)
                break

        else:
            if ready:
                self.update('ready', context)

    def update(self, new_status, context, new_output=None):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        if not isinstance(new_status, StepStateStatus):
            new_status = getattr(StepStateStatus, new_status)

        _LOGGER.info('Updating step %r status: %s -> %s',
                     self.name, self.status.name, new_status.name)
        self.status = new_status

        if self.status is StepStateStatus.ready:
            self.input = self._prepare()
        elif self.status is StepStateStatus.running:
            pass
        elif self.is_completed:
            self._record(new_output)
            for child in self.children:
                child.check_requirements(context)
        elif self.status is StepStateStatus.aborted:
            # The workflow will abort, nothing else to do
            pass
        else:
            # FIXME: cleanup

            raise Exception('Invalid update')

        # Record change in history
        self.history.append((self.status, context))

    def run(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        return self.step.run(self.input)


1			from __future__ import (
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This module should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
2			absolute_import,
3			division,
4			print_function
5			)
6
7			import weakref
8			import logging
9			import collections
10
11			from .state_status import (
12			StateStatus,
13			StepStateStatus,
14			)
15			from .step import Step
16
17			_LOGGER = logging.getLogger(__name__)
18			INIT_STEP = '$init'
19			END_STEP = '$end'
20
21
22			class DeciderStepResult(object):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
23			pass
24
25
26			class DeciderStep(Step):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
27			def run(self, _step_input):
28			return DeciderStepResult()
29
30			def prepare(self, _context):
31			return None
32
33			def render(self, output):
34			# The out of the INIT step is the input to the workflow
35			return output
36
37
38			class State(object):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
39			__slots__ = (
40			'status',
41			'step_states',
42			'_context',
43			'_orphaned_steps',
44			'_init_step',
45			'_end_step',
46			)
47
48			def __init__(self):
49			"""Init state"""
50			self.status = StateStatus.init
51			self.step_states = {}
52			self._context = None
53			self._orphaned_steps = {}
54			# Add an INIT/END fake steps
55			self._init_step = StepState(
56			step=DeciderStep(INIT_STEP),
57			status='running',
58			context='__init__'
59			)
60			self._end_step = StepState(
61			step=DeciderStep(END_STEP,
62			requires=[(INIT_STEP, 'completed')]),
63			context='__init__'
64			)
65			self._stepstate_insert(self._init_step)
66			self._stepstate_insert(self._end_step)
67
68			def __repr__(self):
69			return '{ctype}(status={status},steps={steps})'.format(
70			ctype=self.__class__.__name__,
71			status=self.status.name,
72			steps=len(self.step_states)
73			)
74
75			#################################################
76			def is_in_state(self, state_name):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
77			desired_state = getattr(StateStatus, state_name)
78			return self.status.means(desired_state)
79
80			#################################################
81			# Context management
82			def __call__(self, context):
83			self._context = context
84			return self
85
86			def __enter__(self):
87			assert self._context is not None
88
89			def __exit__(self, exc_type, exc_value, traceback):
90			assert self._context is not None
91			# Only clear the context is we didn't encounter an exception. Otherwise
92			# keep it for debug purposes
93			if exc_type is None:
94			self._context = None
95
96			#################################################
97			# State manipulations
98			def set_abort(self):
99			"""Abort the state machine.
100			"""
101			assert self._context is not None
102			self.step_update(END_STEP, 'aborted')
103
104			def set_input(self, input_data):
105			"""Set the input of the state machine.
106			"""
107			assert self._context is not None
108			assert self.status is StateStatus.init
109
110			self.step_update(INIT_STEP, 'completed', new_data=input_data)
111			self.status = StateStatus.running
112
113			def step_update(self, step_name, new_status, new_data=None):
114			"""Update a Step with new status and, optionally, output data.
115			"""
116			assert self._context is not None
117			step_state = self.step_states[step_name]
118			step_state.update(new_status,
119			context=self._context,
120			new_output=new_data)
121
122			if self._end_step.status is StepStateStatus.ready:
123			self.status = StateStatus.succeeded
124			elif self._end_step.status is StepStateStatus.aborted:
125			self.status = StateStatus.failed
126
127			def step_insert(self, step):
128			"""Add a step definition to the state.
129			"""
130			assert self._context is not None
131			step_state = StepState(step, self._context)
132			_LOGGER.info('Defining new step %r in state', step_state)
133
134			# All steps are children of the root INIT_STEP step
135			self._init_step.children.add(step_state)
136			step_state.parents.add(self._init_step)
137
138			if self._stepstate_insert(step_state):
139			# All steps are a parent of END_STEP
140			# FIXME: Could be optimized so that only steps without children are
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
141			# parented to END_STEP
142			self._end_step.parents.add(step_state)
143			self._end_step.step.requires[step_state.step.name] = \
144			StepStateStatus.completed
145
146			step_state.children.add(self._end_step)
147			# See if we can re-parents some previously orphaned Step with the
148			# one we have just added.
149			orphans = self._orphaned_steps.pop(step.name, [])
150			for orphan in orphans:
151			self._stepstate_insert(orphan)
152
153			#################################################
154			def step_next(self, hint=None):
155			"""Search for steps ready to be ran.
156
157			:param str hint:
158			Place to start searching steps that are now ready to be scheduled
159			(usually, the last `completed` step).
160			"""
161			ready_steps = set()
162			# Were we given a place to start looking?
163			if hint:
164			hint_step = self.step_states[hint]
165
166			# Check that this is completed
167			if not hint_step.is_completed:
168			# FIXME: This should be an error
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
169			return set()
170
171			for child in hint_step.children:
172			if child.status is StepStateStatus.ready:
173			ready_steps.add(child)
174
175			else:
176			# Walk the whole tree collecting ready StepState
177			for child in self._init_step.children:
178			if child.status is StepStateStatus.ready:
179			ready_steps.add(child)
180			else:
181			child_name = child.step.name
182			ready_steps \|= self.step_next(child_name)
183
184			return ready_steps
185
186			def _stepstate_insert(self, step_state):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
187			if not step_state.step.requires:
188			self.step_states[step_state.name] = step_state
189			return True
190
191			elif all([required in self.step_states
192			for required in step_state.step.requires.keys()]):
193			# All the required steps are defined, update their children set and
194			# record them in this step_state's parents set
195			for required in step_state.step.requires.keys():
196			self.step_states[required].children.add(step_state)
197			step_state.parents.add(self.step_states[required])
198
199			self.step_states[step_state.name] = step_state
200			return True
201
202			else:
203			# We failed to insert this step_state
204			# Parent steps are missing, set it as orphaned
205			for required in step_state.step.requires.keys():
206			if required not in self.step_states:
207			self._orphaned_steps.setdefault(
208			required, set()
209			).add(step_state)
210			return False
211
212
213			class StepState(object):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
214
215			__slots__ = ('status',
216			'step',
217			'input',
218			'output',
219			'children',
220			'parents',
221			'history',
222			'__weakref__')
223
224			def __init__(self, step, context,
225			status=StepStateStatus.pending):
226
227			if not isinstance(status, StepStateStatus):
228			status = getattr(StepStateStatus, status)
229
230			self.step = step
231			self.status = status
232			self.input = None
233			self.output = None
234			self.children = weakref.WeakSet()
235			self.parents = weakref.WeakSet()
236			self.history = collections.deque()
237			self.history.append(
238			(status, context)
239			)
240
241			def __repr__(self):
242			return 'StepState({name}:{status})'.format(
243			name=self.step.name, status=self.status.name
244			)
245
246			@property
247			def name(self):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
248			return self.step.name
249
250			@property
251			def is_completed(self):
252			"""`True` if the status is either `succeeded`, `failed` or `skipped`.
253			"""
254			return self.status.means(StepStateStatus.completed)
255
256			def _prepare(self):
257			"""Prepare the input from the step input template and the parents data.
258			"""
259			assert self.status is StepStateStatus.ready
260			context = self._build_context()
261			render = self.step.prepare(context)
262			return render
263
264			def _build_context(self):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
265			context = {}
266			for parent in self.parents:
267			if parent.name is INIT_STEP:
268			context.update({'__input__': parent.output})
269			else:
270			context.update({parent.name: parent.output})
271			return context
272
273			def _record(self, output):
274			"""Invoke the step rendering of the results."""
275			attrs = self.step.render(output)
276			self.output = attrs
277
278			def check_requirements(self, context):
279			"""`True` if the Step is ready to be evaluated.
280
281			This means the step hasn't ran yet and all its parents are completed.
282			"""
283			if self.status is StepStateStatus.ready:
284			return
285			# You cannot be ready if you are already completed/aborted
286			if self.status is not StepStateStatus.pending:
287			return
288
289			# Check that all our parents are completed per the step's requirements.
290			_LOGGER.info('Step %r requirements %r', self, self.step.requires)
291
292			ready = True
293			for parent in self.parents:
294			if parent.name is INIT_STEP:
295			continue
296			assert parent.name in self.step.requires
297
298			if not parent.is_completed:
299			ready = False
300			continue
301
302			req_status = self.step.requires[parent.name]
303			_LOGGER.info('Checking parent %r meets requirement %r',
304			parent, req_status)
305
306			if not parent.status.means(req_status):
307			self.update('skipped', context)
308			break
309
310			else:
311			if ready:
312			self.update('ready', context)
313
314			def update(self, new_status, context, new_output=None):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
315			if not isinstance(new_status, StepStateStatus):
316			new_status = getattr(StepStateStatus, new_status)
317
318			_LOGGER.info('Updating step %r status: %s -> %s',
319			self.name, self.status.name, new_status.name)
320			self.status = new_status
321
322			if self.status is StepStateStatus.ready:
323			self.input = self._prepare()
324			elif self.status is StepStateStatus.running:
325			pass
326			elif self.is_completed:
327			self._record(new_output)
328			for child in self.children:
329			child.check_requirements(context)
330			elif self.status is StepStateStatus.aborted:
331			# The workflow will abort, nothing else to do
332			pass
333			else:
334			# FIXME: cleanup
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
335			raise Exception('Invalid update')
336
337			# Record change in history
338			self.history.append((self.status, context))
339
340			def run(self):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
341			return self.step.run(self.input)
342

sportarchive / CloudProcessingEngine-Decider

StepState.check_requirements() F last analyzed 2016-11-10 17:43 UTC

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

StepState.check_requirements() F
last analyzed 2016-11-10 17:43 UTC