Source code for edgy.workflow.workflow

# -*- coding: utf-8 -*-
from collections import OrderedDict

import six

from edgy.workflow.transition import Transition
from edgy.workflow.constants import WILDCARD

class DeclarativeTransitionsMetaclass(type):
    Metaclass that collects Transitions declared on the base classes.

    def __new__(mcs, name, bases, attrs):
        # Collect transitions from current class.
        current_transitions = []
        for key, value in list(attrs.items()):
            if isinstance(value, Transition):
                current_transitions.append((key, value))
        current_transitions.sort(key=lambda x: x[1].creation_counter)
        attrs['declared_transitions'] = OrderedDict(current_transitions)

        new_class = (super(DeclarativeTransitionsMetaclass, mcs).__new__(mcs, name, bases, attrs))

        # Walk through the MRO.
        declared_transitions = OrderedDict()
        for base in reversed(new_class.__mro__):
            # Collect transitions from base class.
            if hasattr(base, 'declared_transitions'):

            # Field shadowing.
            for attr, value in base.__dict__.items():
                if value is None and attr in declared_transitions:

        new_class.declared_transitions = declared_transitions

        return new_class

[docs]class Workflow(six.with_metaclass(DeclarativeTransitionsMetaclass)): """ A ``Workflow`` is a coherent set of Transitions meant to define a state machine system. """ @property def states(self): """ Set of valid known states for this workflow. Beware, if you're using wildcard as source, there can be states you expect as valid that this instance does not know about, and will treat as invalid. :return: set[str] """ return set(self._valid_states) @property def transitions(self): """ Set of transitions living in this state machine system. :return: set[edgy.workflow.Transition] """ return set(self._transitions.items()) def __init__(self): # states and transitions indexes self._transitions = OrderedDict() self._transitions_by_source = {} self._valid_states = {WILDCARD} for name, transition in self.declared_transitions.items(): self.add_transition(transition, name=name) def __contains__(self, item): return item in self._transitions def __getitem__(self, item): return self._transitions[item]
[docs] def add_transition(self, transition, name=None): """Add a transition to this workflow instance, to be used on stateful subjects later. :param edgy.workflow.Transition transition: Transition to add. """ name = name or transition.__name__ # store the transition by name self._transitions[name] = transition # ensure we know source and target states as valid self._valid_states = self._valid_states.union(set(transition.source)) self._valid_states.add( # index transitions by source state for source_state in transition.source: if not source_state in self._transitions_by_source: self._transitions_by_source[source_state] = {} self._transitions_by_source[source_state][name] = transition return transition
def get_available_transitions_for(self, subject): transitions = self._transitions_by_source.get(subject.state, {}) transitions.update(self._transitions_by_source.get(WILDCARD, {})) return transitions