# -*- coding: utf-8 -*-
# Copyright (c) 2011-2013 Raphaël Barrois
# This code is distributed under the two-clause BSD License.
"""Base components of XWorkflows."""
import logging
import re
import warnings
from .compat import is_python3, is_string, u
from . import utils
[docs]class WorkflowError(Exception):
"""Base class for errors from the xworkflows module."""
[docs]class AbortTransition(WorkflowError):
"""Raised to prevent a transition from proceeding."""
[docs]class InvalidTransitionError(AbortTransition):
"""Raised when trying to perform a transition not available from current state."""
[docs]class ForbiddenTransition(AbortTransition):
"""Raised when the 'check' hook of a transition was defined and returned False."""
class State(object):
"""A state within a workflow.
Attributes:
name (str): the name of the state
title (str): the human-readable title for the state
"""
STATE_NAME_RE = re.compile(r'\w+$')
def __init__(self, name, title):
if not self.STATE_NAME_RE.match(name):
raise ValueError('Invalid state name %s.' % name)
self.name = name
self.title = title
def __str__(self):
return self.name
def __repr__(self):
return '<%s: %r>' % (self.__class__.__name__, self.name)
class StateList(object):
"""A list of states."""
def __init__(self, states):
self._states = dict((st.name, st) for st in states)
self._order = tuple(st.name for st in states)
def __getattr__(self, name):
try:
return self._states[name]
except KeyError:
raise AttributeError('StateList %s has no state named %s' % (self, name))
def __len__(self):
return len(self._states)
def __getitem__(self, name_or_state):
if isinstance(name_or_state, State):
return self._states[name_or_state.name]
else:
return self._states[name_or_state]
def __repr__(self):
return '%s(%r)' % (self.__class__.__name__, self._states)
def __iter__(self):
for name in self._order:
yield self._states[name]
def __contains__(self, state):
if isinstance(state, State):
return state.name in self._states and self._states[state.name] == state
else: # Expect a state name
return state in self._states
class Transition(object):
"""A transition.
Attributes:
name (str): the name of the Transition
source (State list): the 'source' states of the transition
target (State): the 'target' state of the transition
"""
def __init__(self, name, source, target):
self.name = name
if isinstance(source, State):
source = [source]
self.source = source
self.target = target
def __repr__(self):
return '%s(%r, %r, %r)' % (self.__class__.__name__,
self.name, self.source, self.target)
class TransitionList(object):
"""Holder for the transitions of a given workflow."""
def __init__(self, transitions):
"""Create a TransitionList.
Args:
transitions (list of (name, source, target) tuple): the transitions
to include.
"""
self._transitions = {}
self._order = []
for trdef in transitions:
self._transitions[trdef.name] = trdef
self._order.append(trdef.name)
def __len__(self):
return len(self._transitions)
def __getattr__(self, name):
try:
return self._transitions[name]
except KeyError:
raise AttributeError(
"TransitionList %s has no transition named %s." % (self, name))
def __getitem__(self, name):
return self._transitions[name]
def __iter__(self):
for name in self._order:
yield self._transitions[name]
def __contains__(self, value):
if isinstance(value, Transition):
return value.name in self._transitions and self._transitions[value.name] == value
else:
return value in self._transitions
def available_from(self, state):
"""Retrieve all transitions available from a given state.
Args:
state (State): the initial state.
Yields:
Transition: all transitions starting from that state
"""
for transition in self:
if state in transition.source:
yield transition
def __repr__(self):
return '%s(%r)' % (self.__class__.__name__, self._transitions.values())
def _setup_states(state_definitions, prev=()):
"""Create a StateList object from a 'states' Workflow attribute."""
states = list(prev)
for state_def in state_definitions:
if len(state_def) != 2:
raise TypeError(
"The 'state' attribute of a workflow should be "
"a two-tuple of strings; got %r instead." % (state_def,)
)
name, title = state_def
state = State(name, title)
if any(st.name == name for st in states):
# Replacing an existing state
states = [state if st.name == name else st for st in states]
else:
states.append(state)
return StateList(states)
def _setup_transitions(tdef, states, prev=()):
"""Create a TransitionList object from a 'transitions' Workflow attribute.
Args:
tdef: list of transition definitions
states (StateList): already parsed state definitions.
prev (TransitionList): transition definitions from a parent.
Returns:
TransitionList: the list of transitions defined in the 'tdef' argument.
"""
trs = list(prev)
for transition in tdef:
if len(transition) == 3:
(name, source, target) = transition
if is_string(source) or isinstance(source, State):
source = [source]
source = [states[src] for src in source]
target = states[target]
tr = Transition(name, source, target)
else:
raise TypeError(
"Elements of the 'transition' attribute of a "
"workflow should be three-tuples; got %r instead." % (transition,)
)
if any(prev_tr.name == tr.name for prev_tr in trs):
# Replacing an existing state
trs = [tr if prev_tr.name == tr.name else prev_tr for prev_tr in trs]
else:
trs.append(tr)
return TransitionList(trs)
HOOK_BEFORE = 'before'
HOOK_AFTER = 'after'
HOOK_CHECK = 'check'
HOOK_ON_ENTER = 'on_enter'
HOOK_ON_LEAVE = 'on_leave'
class Hook(object):
"""A hook to run when a transition occurs.
Attributes:
kind (str): the kind of hook
priority (int): the priority of the hook (higher values run first)
function (callable): the actual function to call
names (str list): name of the transitions or states to which the hook
relates. The special value '*' means 'applies to all transitions/
states'.
Hooks are sortable by descending priority and ascending function name.
Hook kinds are as follow:
- HOOK_BEFORE: run before the related transitions
- HOOK_AFTER: run after the related transitions
- HOOK_CHECK: run as part of pre-transition checks (return value matters)
- HOOK_ON_ENTER: run just after a transition entering a related state
- HOOK_ON_LEAVE: run just before a transition leaving from a related state
"""
def __init__(self, kind, function, *names, **kwargs):
assert kind in (
HOOK_BEFORE, HOOK_AFTER, HOOK_CHECK,
HOOK_ON_ENTER, HOOK_ON_LEAVE)
self.kind = kind
self.priority = kwargs.get('priority', 0)
self.function = function
self.names = names or ('*',)
def _match_state(self, state):
"""Checks whether a given State matches self.names."""
return (self.names == '*'
or state in self.names
or state.name in self.names)
def _match_transition(self, transition):
"""Checks whether a given Transition matches self.names."""
return (self.names == '*'
or transition in self.names
or transition.name in self.names)
def applies_to(self, transition, from_state=None):
"""Whether this hook applies to the given transition/state.
Args:
transition (Transition): the transition to check
from_state (State or None): the state to check. If absent, the check
is 'might this hook apply to the related transition, given a
valid source state'.
"""
if '*' in self.names:
return True
elif self.kind in (HOOK_BEFORE, HOOK_AFTER, HOOK_CHECK):
return self._match_transition(transition)
elif self.kind == HOOK_ON_ENTER:
return self._match_state(transition.target)
elif from_state is None:
# Testing whether the hook may apply to at least one source of the
# transition
return any(self._match_state(src) for src in transition.source)
else:
return self._match_state(from_state)
def __call__(self, *args, **kwargs):
return self.function(*args, **kwargs)
def __eq__(self, other):
"""Equality is based on priority, function and kind."""
if not isinstance(other, Hook):
return NotImplemented
return (
self.priority == other.priority
and self.function == other.function
and self.kind == other.kind
and self.names == other.names
)
def __ne__(self, other):
if not isinstance(other, Hook):
return NotImplemented
return not (self == other)
def __lt__(self, other):
"""Compare hooks of the same kind."""
if not isinstance(other, Hook):
return NotImplemented
return (
(other.priority, self.function.__name__)
< (self.priority, other.function.__name__))
def __gt__(self, other):
"""Compare hooks of the same kind."""
if not isinstance(other, Hook):
return NotImplemented
return (
(other.priority, self.function.__name__)
> (self.priority, other.function.__name__))
def __repr__(self):
return '<%s: %s %r>' % (
self.__class__.__name__, self.kind, self.function)
class ImplementationWrapper(object):
"""Wraps a transition implementation.
Emulates a function behaviour, but provides a few extra features.
Attributes:
instance (WorkflowEnabled): the instance to update
.
field_name (str): the name of the field of the instance to update.
transition (Transition): the transition to perform
workflow (Workflow): the workflow to which this is related.
hooks (Hook list): optional hooks to call during the transition
implementation (callable): the code to invoke between 'before' and the
state update.
"""
def __init__(self, instance, field_name, transition, workflow,
implementation, hooks=None):
self.instance = instance
self.field_name = field_name
self.transition = transition
self.workflow = workflow
self.hooks = hooks or {}
self.implementation = implementation
self.__doc__ = implementation.__doc__
@property
def current_state(self):
return getattr(self.instance, self.field_name)
def _pre_transition_checks(self):
"""Run the pre-transition checks."""
current_state = getattr(self.instance, self.field_name)
if current_state not in self.transition.source:
raise InvalidTransitionError(
"Transition '%s' isn't available from state '%s'." %
(self.transition.name, current_state.name))
for check in self._filter_hooks(HOOK_CHECK):
if not check(self.instance):
raise ForbiddenTransition(
"Transition '%s' was forbidden by "
"custom pre-transition check." % self.transition.name)
def _filter_hooks(self, *hook_kinds):
"""Filter a list of hooks, keeping only applicable ones."""
hooks = sum((self.hooks.get(kind, []) for kind in hook_kinds), [])
return sorted(hook for hook in hooks
if hook.applies_to(self.transition, self.current_state))
def _pre_transition(self, *args, **kwargs):
for hook in self._filter_hooks(HOOK_BEFORE, HOOK_ON_LEAVE):
hook(self.instance, *args, **kwargs)
def _during_transition(self, *args, **kwargs):
return self.implementation(self.instance, *args, **kwargs)
def _log_transition(self, from_state, *args, **kwargs):
self.workflow.log_transition(
self.transition, from_state, self.instance,
*args, **kwargs)
def _post_transition(self, result, *args, **kwargs):
"""Performs post-transition actions."""
for hook in self._filter_hooks(HOOK_AFTER, HOOK_ON_ENTER):
hook(self.instance, result, *args, **kwargs)
def __call__(self, *args, **kwargs):
"""Run the transition, with all checks."""
self._pre_transition_checks()
# Call hooks.
self._pre_transition(*args, **kwargs)
result = self._during_transition(*args, **kwargs)
from_state = getattr(self.instance, self.field_name)
setattr(self.instance, self.field_name, self.transition.target)
# Call hooks.
self._log_transition(from_state, *args, **kwargs)
self._post_transition(result, *args, **kwargs)
return result
def is_available(self):
"""Check whether this transition is available on the current object.
Returns:
bool
"""
try:
self._pre_transition_checks()
except (InvalidTransitionError, ForbiddenTransition):
return False
return True
def __repr__(self):
return "<%s for %r on %r: %r>" % (
self.__class__.__name__,
self.transition.name, self.field_name, self.implementation)
class ImplementationProperty(object):
"""Holds an implementation of a transition.
This class is a 'non-data descriptor', somewhat similar to a property.
Attributes:
field_name (str): the name of the field of the instance to update.
transition (Transition): the transition to perform
workflow (Workflow): the workflow to which this is related.
hooks (Hook list): hooks to apply along with the transition.
implementation (callable): the code to invoke between 'before' and the
state update.
"""
def __init__(self, field_name, transition, workflow, implementation, hooks=None):
self.field_name = field_name
self.transition = transition
self.workflow = workflow
self.hooks = hooks or {}
self.implementation = implementation
self.__doc__ = implementation.__doc__
def copy(self):
return self.__class__(
field_name=self.field_name,
transition=self.transition,
workflow=self.workflow,
implementation=self.implementation,
# Don't copy hooks: they'll be re-generated during metaclass __new__
hooks={},
)
def add_hook(self, hook):
self.hooks.setdefault(hook.kind, []).append(hook)
def __get__(self, instance, owner):
if instance is None:
return self
if not isinstance(instance, BaseWorkflowEnabled):
raise TypeError(
"Unable to apply transition '%s' to object %r, which is not "
"attached to a Workflow." % (self.transition.name, instance))
return self.workflow.implementation_class(
instance,
self.field_name, self.transition, self.workflow,
self.implementation, self.hooks)
def __repr__(self):
return "<%s for '%s' on '%s': %s>" % (
self.__class__.__name__,
self.transition.name, self.field_name, self.implementation)
class TransitionWrapper(object):
"""Mark that a method should be used for a transition with a different name.
Attributes:
trname (str): the name of the transition that the method implements
func (function): the decorated method
"""
def __init__(self, trname, field='', check=None, before=None, after=None):
self.trname = trname
self.field = field
self.check = check
self.before = before
self.after = after
self.func = None
def __call__(self, func):
self.func = func
if self.trname == '':
self.trname = func.__name__
return self
def __repr__(self):
return "<%s for %r: %s>" % (self.__class__.__name__, self.trname, self.func)
[docs]def transition(trname='', field='', check=None, before=None, after=None):
"""Decorator to declare a function as a transition implementation."""
if callable(trname):
raise ValueError(
"The @transition decorator should be called as "
"@transition(['transition_name'], **kwargs)")
if check or before or after:
warnings.warn(
"The use of check=, before= and after= in @transition decorators is "
"deprecated in favor of @transition_check, @before_transition and "
"@after_transition decorators.",
DeprecationWarning,
stacklevel=2)
return TransitionWrapper(trname, field=field, check=check, before=before, after=after)
def _make_hook_dict(fun):
"""Ensure the given function has a xworkflows_hook attribute.
That attribute has the following structure:
>>> {
... 'before': [('state', <TransitionHook>), ...],
... }
"""
if not hasattr(fun, 'xworkflows_hook'):
fun.xworkflows_hook = {
HOOK_BEFORE: [],
HOOK_AFTER: [],
HOOK_CHECK: [],
HOOK_ON_ENTER: [],
HOOK_ON_LEAVE: [],
}
return fun.xworkflows_hook
class _HookDeclaration(object):
"""Base class for decorators declaring methods as transition hooks.
Args:
*names (str tuple): name of the states/transitions to bind to; use '*'
for 'all'
priority (int): priority of the hook, defaults to 0
field (str): name of the field to which the hooked transition relates
Usage:
>>> @_HookDeclaration('foo', 'bar', priority=4)
... def my_hook(self):
... pass
"""
def __init__(self, *names, **kwargs):
if not names:
names = ('*',)
self.names = names
self.priority = kwargs.get('priority', 0)
self.field = kwargs.get('field', '')
def _as_hook(self, func):
return Hook(self.hook_name, func, *self.names, priority=self.priority)
def __call__(self, func):
hook_dict = _make_hook_dict(func)
hooks = hook_dict[self.hook_name]
hooks.append((self.field, self._as_hook(func)))
return func
[docs]class before_transition(_HookDeclaration):
"""Decorates a method that should be called before a given transition.
Example:
>>> @before_transition('foobar')
... def blah(self):
... pass
"""
hook_name = HOOK_BEFORE
[docs]class after_transition(_HookDeclaration):
"""Decorates a method that should be called after a given transition.
Example:
>>> @after_transition('foobar')
... def blah(self):
... pass
"""
hook_name = HOOK_AFTER
[docs]class transition_check(_HookDeclaration):
"""Decorates a method that should be called after a given transition.
Example:
>>> @transition_check('foobar')
... def blah(self):
... pass
"""
hook_name = HOOK_CHECK
[docs]class on_enter_state(_HookDeclaration):
"""Decorates a method that should be used as a hook for a state.
Example:
>>> @on_enter_state('foo', 'bar')
... def blah(self):
... pass
"""
hook_name = HOOK_ON_ENTER
[docs]class on_leave_state(_HookDeclaration):
"""Decorates a method that should be used as a hook for a state.
Example:
>>> @on_leave_state('foo', 'bar')
... def blah(self):
... pass
"""
hook_name = HOOK_ON_LEAVE
def noop(instance, *args, **kwargs):
"""NoOp function, ignores all arguments."""
pass
class ImplementationList(object):
"""Stores all implementations.
Attributes:
state_field (str): the name of the field holding the state of objects.
implementations (dict(str => ImplementationProperty)): maps a transition
name to the associated implementation.
workflow (Workflow): the related workflow
transitions_at (dict(str => str)): maps a transition name to the
name of the attribute holding the related implementation.
custom_implems (str set): list of transition names for which a custom
implementation has been defined.
"""
def __init__(self, state_field, workflow):
self.state_field = state_field
self.workflow = workflow
self.implementations = {}
self.transitions_at = {}
self.custom_implems = set()
def load_parent_implems(self, parent_implems):
"""Import previously defined implementations.
Args:
parent_implems (ImplementationList): List of implementations defined
in a parent class.
"""
for trname, attr, implem in parent_implems.get_custom_implementations():
self.implementations[trname] = implem.copy()
self.transitions_at[trname] = attr
self.custom_implems.add(trname)
def add_implem(self, transition, attribute, function, **kwargs):
"""Add an implementation.
Args:
transition (Transition): the transition for which the implementation
is added
attribute (str): the name of the attribute where the implementation
will be available
function (callable): the actual implementation function
**kwargs: extra arguments for the related ImplementationProperty.
"""
implem = ImplementationProperty(
field_name=self.state_field,
transition=transition,
workflow=self.workflow,
implementation=function,
**kwargs)
self.implementations[transition.name] = implem
self.transitions_at[transition.name] = attribute
return implem
def should_collect(self, value):
"""Decide whether a given value should be collected."""
return (
# decorated with @transition
isinstance(value, TransitionWrapper)
# Relates to a compatible transition
and value.trname in self.workflow.transitions
# Either not bound to a state field or bound to the current one
and (not value.field or value.field == self.state_field))
def collect(self, attrs):
"""Collect the implementations from a given attributes dict."""
for name, value in attrs.items():
if self.should_collect(value):
transition = self.workflow.transitions[value.trname]
if (
value.trname in self.implementations
and value.trname in self.custom_implems
and name != self.transitions_at[value.trname]):
# We already have an implementation registered.
other_implem_at = self.transitions_at[value.trname]
raise ValueError(
"Error for attribute %s: it defines implementation "
"%s for transition %s, which is already implemented "
"at %s." % (name, value, transition, other_implem_at))
implem = self.add_implem(transition, name, value.func)
self.custom_implems.add(transition.name)
if value.check:
implem.add_hook(Hook(HOOK_CHECK, value.check))
if value.before:
implem.add_hook(Hook(HOOK_BEFORE, value.before))
if value.after:
implem.add_hook(Hook(HOOK_AFTER, value.after))
def get_custom_implementations(self):
"""Retrieve a list of cutom implementations.
Yields:
(str, str, ImplementationProperty) tuples: The name of the attribute
an implementation lives at, the name of the related transition,
and the related implementation.
"""
for trname in self.custom_implems:
attr = self.transitions_at[trname]
implem = self.implementations[trname]
yield (trname, attr, implem)
def add_missing_implementations(self):
for transition in self.workflow.transitions:
if transition.name not in self.implementations:
self.add_implem(transition, transition.name, noop)
def register_hooks(self, cls):
for field, value in utils.iterclass(cls):
if callable(value) and hasattr(value, 'xworkflows_hook'):
self.register_function_hooks(value)
def register_function_hooks(self, func):
"""Looks at an object method and registers it for relevent transitions."""
for hook_kind, hooks in func.xworkflows_hook.items():
for field_name, hook in hooks:
if field_name and field_name != self.state_field:
continue
for transition in self.workflow.transitions:
if hook.applies_to(transition):
implem = self.implementations[transition.name]
implem.add_hook(hook)
def _may_override(self, implem, other):
"""Checks whether an ImplementationProperty may override an attribute."""
if isinstance(other, ImplementationProperty):
# Overriding another custom implementation for the same transition
# and field
return (other.transition == implem.transition and other.field_name == self.state_field)
elif isinstance(other, TransitionWrapper):
# Overriding the definition that led to adding the current
# ImplementationProperty.
return (
other.trname == implem.transition.name
and (not other.field or other.field == self.state_field)
and other.func == implem.implementation)
return False
def fill_attrs(self, attrs):
"""Update the 'attrs' dict with generated ImplementationProperty."""
for trname, attrname in self.transitions_at.items():
implem = self.implementations[trname]
if attrname in attrs:
conflicting = attrs[attrname]
if not self._may_override(implem, conflicting):
raise ValueError(
"Can't override transition implementation %s=%r with %r" %
(attrname, conflicting, implem))
attrs[attrname] = implem
return attrs
def transform(self, attrs):
"""Perform all actions on a given attribute dict."""
self.collect(attrs)
self.add_missing_implementations()
self.fill_attrs(attrs)
class WorkflowMeta(type):
"""Base metaclass for all Workflows.
Sets the 'states', 'transitions', and 'initial_state' attributes.
"""
def __new__(mcs, name, bases, attrs):
state_defs = attrs.pop('states', [])
transitions_defs = attrs.pop('transitions', [])
initial_state = attrs.pop('initial_state', None)
new_class = super(WorkflowMeta, mcs).__new__(mcs, name, bases, attrs)
new_class.states = _setup_states(state_defs, getattr(new_class, 'states', []))
new_class.transitions = _setup_transitions(
transitions_defs,
new_class.states,
getattr(new_class, 'transitions', []),
)
if initial_state is not None:
new_class.initial_state = new_class.states[initial_state]
return new_class
class BaseWorkflow(object):
"""Base class for all workflows.
Attributes:
states (StateList): list of states of this Workflow
transitions (TransitionList): list of Transitions of this Workflow
initial_state (State): initial state for the Workflow
implementation_class (ImplementationWrapper subclass): class to use
for transition implementation wrapping.
For each transition, a ImplementationWrapper with the same name (unless
another name has been specified through the use of the @transition
decorator) is provided to perform the specified transition.
"""
implementation_class = ImplementationWrapper
def log_transition(self, transition, from_state, instance, *args, **kwargs):
"""Log a transition.
Args:
transition (Transition): the name of the performed transition
from_state (State): the source state
instance (object): the modified object
Kwargs:
Any passed when calling the transition
"""
logger = logging.getLogger('xworkflows.transitions')
try:
instance_repr = u(repr(instance), 'ignore')
except (UnicodeEncodeError, UnicodeDecodeError):
instance_repr = u("<bad repr>")
logger.info(
u("%s performed transition %s.%s (%s -> %s)"), instance_repr,
self.__class__.__name__, transition.name, from_state.name,
transition.target.name)
# Workaround for metaclasses on python2/3.
# Equivalent to:
# Python2
#
# class Workflow(BaseWorkflow):
# __metaclass__ = WorkflowMeta
#
# Python3
#
# class Workflow(metaclass=WorkflowMeta):
# pass
Workflow = WorkflowMeta('Workflow', (BaseWorkflow,), {})
class StateWrapper(object):
"""Slightly enhanced wrapper around a base State object.
Knows about the workflow.
"""
def __init__(self, state, workflow):
self.state = state
self.workflow = workflow
for st in workflow.states:
setattr(self, 'is_%s' % st.name, st.name == self.state.name)
def __eq__(self, other):
if isinstance(other, self.__class__):
return self.state == other.state
elif isinstance(other, State):
return self.state == other
elif is_string(other):
return self.state.name == other
else:
return NotImplemented
def __ne__(self, other):
return not (self == other)
def __str__(self):
return self.state.name
def __repr__(self):
return '<%s: %r>' % (self.__class__.__name__, self.state)
def __getattr__(self, attr):
if attr == 'state':
raise AttributeError(
'Trying to access attribute %s of a non-initialized %r object!'
% (attr, self.__class__))
else:
return getattr(self.state, attr)
if not is_python3:
def __unicode__(self):
return u(str(self))
def __hash__(self):
# A StateWrapper should compare equal to its name.
return hash(self.state.name)
def transitions(self):
"""Retrieve a list of transitions available from this state."""
return self.workflow.transitions.available_from(self.state)
class StateProperty(object):
"""Property-like attribute holding the state of a WorkflowEnabled object.
The state is stored in the internal __dict__ of the instance.
"""
def __init__(self, workflow, state_field_name):
super(StateProperty, self).__init__()
self.workflow = workflow
self.field_name = state_field_name
def __get__(self, instance, owner):
"""Retrieve the current state of the 'instance' object."""
if instance is None:
return self
state = instance.__dict__.get(self.field_name,
self.workflow.initial_state)
return StateWrapper(state, self.workflow)
def __set__(self, instance, value):
"""Set the current state of the 'instance' object."""
try:
state = self.workflow.states[value]
except KeyError:
raise ValueError("Value %s is not a valid state for workflow %s." % (value, self.workflow))
instance.__dict__[self.field_name] = state
def __str__(self):
return 'StateProperty(%s, %s)' % (self.workflow, self.field_name)
class StateField(object):
"""Indicates that a given class attribute is actually a workflow state."""
def __init__(self, workflow):
self.workflow = workflow
class WorkflowEnabledMeta(type):
"""Base metaclass for all Workflow Enabled objects.
Defines:
- one class attribute for each the attached workflows,
- a '_workflows' attribute, a dict mapping each field_name to the related
Workflow,
- a '_xworkflows_implems' attribute, a dict mapping each field_name to a
dict of related ImplementationProperty.
- one class attribute for each transition for each attached workflow
"""
@classmethod
def _add_workflow(mcs, field_name, state_field, attrs):
"""Attach a workflow to the attribute list (create a StateProperty)."""
attrs[field_name] = StateProperty(state_field.workflow, field_name)
@classmethod
def _find_workflows(mcs, attrs):
"""Finds all occurrences of a workflow in the attributes definitions.
Returns:
dict(str => StateField): maps an attribute name to a StateField
describing the related Workflow.
"""
workflows = {}
for attribute, value in attrs.items():
if isinstance(value, Workflow):
workflows[attribute] = StateField(value)
return workflows
@classmethod
def _add_transitions(mcs, field_name, workflow, attrs, implems=None):
"""Collect and enhance transition definitions to a workflow.
Modifies the 'attrs' dict in-place.
Args:
field_name (str): name of the field transitions should update
workflow (Workflow): workflow we're working on
attrs (dict): dictionary of attributes to be updated.
implems (ImplementationList): Implementation list from parent
classes (optional)
Returns:
ImplementationList: The new implementation list for this field.
"""
new_implems = ImplementationList(field_name, workflow)
if implems:
new_implems.load_parent_implems(implems)
new_implems.transform(attrs)
return new_implems
@classmethod
def _register_hooks(mcs, cls, implems):
for implem_list in implems.values():
implem_list.register_hooks(cls)
def __new__(mcs, name, bases, attrs):
# Map field_name => StateField
workflows = {}
# Map field_name => ImplementationList
implems = {}
# Collect workflows and implementations from parents
for base in reversed(bases):
if hasattr(base, '_workflows'):
workflows.update(base._workflows)
implems.update(base._xworkflows_implems)
workflows.update(mcs._find_workflows(attrs))
# Update attributes with workflow descriptions, and collect
# implementation declarations.
for field, state_field in workflows.items():
mcs._add_workflow(field, state_field, attrs)
implems[field] = mcs._add_transitions(
field, state_field.workflow, attrs, implems.get(field))
# Set specific attributes for children
attrs['_workflows'] = workflows
attrs['_xworkflows_implems'] = implems
cls = super(WorkflowEnabledMeta, mcs).__new__(mcs, name, bases, attrs)
mcs._register_hooks(cls, implems)
return cls
class BaseWorkflowEnabled(object):
"""Base class for all objects using a workflow.
Attributes:
workflows (dict(str, StateField)): Maps the name of a 'state_field' to
the related Workflow
"""
# Workaround for metaclasses on python2/3.
# Equivalent to:
# Python2
#
# class WorkflowEnabled(BaseWorkflowEnabled):
# __metaclass__ = WorkflowEnabledMeta
#
# Python3
#
# class WorkflowEnabled(metaclass=WorkflowEnabledMeta):
# pass
WorkflowEnabled = WorkflowEnabledMeta('WorkflowEnabled', (BaseWorkflowEnabled,), {})