Internals

This document presents the various classes and components of XWorkflows.

Note

All objects defined in the base module should be considered internal API and subject to change without notice.

Public API consists of the public methods and attributes of the following objects:

Exceptions

The xworkflows module exposes a few specific exceptions:

exception xworkflows.WorkflowError[source]

This is the base for all exceptions from the xworkflows module.

exception xworkflows.AbortTransition(WorkflowError)[source]

This error is raised whenever a transition call fails, either due to state validation or pre-transition checks.

exception xworkflows.InvalidTransitionError(AbortTransition)[source]

This exception is raised when trying to perform a transition from an incompatible state.

exception xworkflows.ForbiddenTransition(AbortTransition)[source]

This exception will be raised when the check parameter of the transition() decorator returns a non-True value.

States

States may be represented with different objects:

The State class

class base.State(name, title)

This class describes a state in the most simple manner: with an internal name and a human-readable title.

name

The name of the State; used as an internal representation of the state, this should only contain ascii letters and numbers.

title

The title of the State; used for display to users.

The StateWrapper class

class base.StateWrapper(state, workflow)

Intended for use as a WorkflowEnabled attribute, this wraps a State with knowledge about the related Workflow.

Its __hash__ is computed from the related name. It compares equal to:

state

The wrapped State

workflow

The Workflow to which this State belongs.

transitions()
Returns:A list of Transition with this State as source

The StateProperty class

class base.StateProperty(workflow, state_field_name)

Special property-like object (technically a data descriptor), this class controls access to the current State of a WorkflowEnabled object.

It performs the following actions:

  • Checks that any set value is a valid State from the workflow (raises ValueError otherwise)
  • Wraps retrieved values into a StateWrapper
workflow

The Workflow to which the attribute is related

field_name

The name of the attribute wrapped by this StateProperty.

Workflows

A Workflow definition is slightly different from the resulting class.

A few class-level declarations will be converted into advanced objects:

Workflow definition

A Workflow definition must inherit from the Workflow class, or use the base.WorkflowMeta metaclass for proper setup.

Defining states

The list of states should be defined as a list of two-tuples of (name, title):

class MyWorkflow(xworkflows.Workflow):
    states = (
        ('initial', "Initial"),
        ('middle', "Intermediary"),
        ('final', "Final - all is said and done."),
    )

This is converted into a StateList object.

class base.StateList

This class acts as a mixed dictionary/object container of states.

It replaces the states list from the Workflow definition.

__len__()

Returns the number of states in the Workflow

__getitem__()

Allows retrieving a State from its name or from an instance, in a dict-like manner

__getattr__()

Allows retrieving a State from its name, as an attribute of the StateList:

MyWorkflow.states.initial == MyWorkflow.states['initial']
__iter__()

Iterates over the states, in the order they were defined

__contains__()

Tests whether a State instance or its name belong to the Workflow

Defining transitions

At a Workflow level, transition are defined in a list of three-tuples:

  • transition name
  • list of the names of source states for the transition, or name of the source state if unique
  • name of the target State
class MyWorkflow(xworkflows.Workflow):
    transitions = (
        ('advance', 'initial', 'middle'),
        ('end', ['initial', 'middle'], 'final'),
    )

This is converted into a TransitionList object.

class base.TransitionList

This acts as a mixed dictionary/object container of transitions.

It replaces the transitions list from the Workflow definition.

__len__()

Returns the number of transitions in the Workflow

__getitem__()

Allows retrieving a Transition from its name or from an instance, in a dict-like manner

__getattr__()

Allows retrieving a Transition from its name, as an attribute of the TransitionList:

MyWorkflow.transitions.accept == MyWorkflow.transitions['accept']
__iter__()

Iterates over the transitions, in the order they were defined

__contains__()

Tests whether a Transition instance or its name belong to the Workflow

available_from(state)

Retrieve the list of Transition available from the given State.

class base.Transition

Container for a transition.

name

The name of the Transition; should be a valid Python identifier

source

A list of source states for this Transition

target

The target State for this Transition

Workflow attributes

A Workflow should inherit from the Workflow base class, or use the WorkflowMeta metaclass (that builds the states, transitions, initial_state attributes).

class xworkflows.Workflow

This class holds the definition of a workflow.

states

A StateList of all State for this Workflow

transitions

A TransitionList of all Transition for this Workflow

initial_state

The initial State for this Workflow

log_transition(transition, from_state, instance, *args, **kwargs)
Parameters:
  • transition – The Transition just performed
  • from_state – The source State of the instance (before performing a transition)
  • instance – The object undergoing a transition
  • args – All non-keyword arguments passed to the transition implementation
  • kwargs – All keyword arguments passed to the transition implementation

This method allows logging all transitions performed by objects using a given workflow.

The default implementation logs to the logging module, in the base logger.

implementation_class

The class to use when creating ImplementationWrapper for a WorkflowEnabled using this Workflow.

Defaults to ImplementationWrapper.

class base.WorkflowMeta

This metaclass will simply convert the states, transitions and initial_state class attributes into the related StateList, TransitionList and State objects.

During this process, some sanity checks are performed:

Applying workflows

In order to use a Workflow, related objects should inherit from the WorkflowEnabled class.

class xworkflows.WorkflowEnabled

This class will handle all specific setup related to using workflows:

_add_workflow(mcs, field_name, state_field, attrs)

Adds a workflow to the attributes dict of the future class.

Parameters:
  • field_name (str) – Name of the field at which the field holding the current state will live
  • state_field (StateField) – The StateField as returned by _find_workflows()
  • attrs (dict) – Attribute dict of the future class, updated with the new StateProperty.

Note

This method is also an extension point for custom XWorkflow-related libraries.

_find_workflows(mcs, attrs)

Find all workflow definitions in a class attributes dict.

Parameters:attrs (dict) – Attribute dict of the future class
Returns:A dict mapping a field name to a StateField describing parameters for the workflow

Note

This method is also an extension point for custom XWorkflow-related libraries.

_workflows

This class-level attribute holds a dict mapping an attribute to the related Workflow.

Note

This is a private attribute, and may change at any time in the future.

_xworkflows_implems

This class-level attribute holds a dict mapping an attribute to the related implementations.

Note

This is a private attribute, and may change at any time in the future.

class base.WorkflowEnabledMeta

This metaclass handles the parsing of WorkflowEnabled and related magic.

Most of the work is handled by ImplementationList, with one instance handling each Workflow attached to the WorkflowEnabled object.

Customizing transitions

A bare WorkflowEnabled subclass definition will be automatically modified to include “noop” implementations for all transitions from related workflows.

In order to customize this behaviour, one should use the transition() decorator on methods that should be called when performing transitions.

xworkflows.transition([trname='', field='', check=None, before=None, after=None])[source]

Decorates a method and uses it for a given Transition.

Parameters:
  • trname (str) – Name of the transition during which the decorated method should be called. If empty, the name of the decorated method is used.
  • field (str) – Name of the field this transition applies to; useful when two workflows define a transition with the same name.
  • check (callable) –

    An optional function to call before running the transition, with the about-to-be-modified instance as single argument.

    Should return True if the transition can proceed.

    Deprecated since version 0.4.0: Will be removed in 0.5.0; use transition_check() instead.

  • before (callable) –

    An optional function to call after checks and before the actual implementation.

    Receives the same arguments as the transition implementation.

    Deprecated since version 0.4.0: Will be removed in 0.5.0; use before_transition() instead.

  • after (callable) –

    An optional function to call after the transition was performed and logged.

    Receives the instance, the implementation return value and the implementation arguments.

    Deprecated since version 0.4.0: Will be removed in 0.5.0; use after_transition() instead.
class base.TransitionWrapper

Actual class holding all values defined by the transition() decorator.

func

The decorated function, wrapped with a few checks and calls.

Hooks

Hooks are declared through a _HookDeclaration decorator, which attaches a specific xworkflows_hook attribute to the decorated method. Methods with such attribute will be collected into Hook objects containing all useful fields.

Registering hooks

xworkflows._make_hook_dict(function)

Ensures that the given function has a xworkflows_hook attributes, and returns it.

The xworkflows_hook is a dict mapping each hook kind to a list of (field, hook) pairs:

function.xworkflows_hook = {
    HOOK_BEFORE: [('state', <Hook: ...>), ('', <Hook: ...>)],
    HOOK_AFTER: [],
    ...
}

Note

Although the xworkflows_hook is considered a private API, it may become an official extension point in future releases.

class base._HookDeclaration

Base class for hook declaration decorators.

It accepts an (optional) list of transition/state names, and priority / field as keyword arguments:

@_HookDeclaration('foo', 'bar')
@_HookDeclaration(priority=42)
@_HookDeclaration('foo', field='state1')
@_HookDeclaration(priority=42, field='state1')
def hook(self):
    pass
names

List of transition or state names the hook applies to

Type:str list
priority

The priority of the hook

Type:int
field

The name of the StateWrapper field whose transitions the hook applies to

Type:str
_as_hook(self, func)

Create a Hook for the given callable

__call__(self, func)

Create a Hook for the function, and store it in the function’s xworkflows_hook attribute.

xworkflows.before_transition(*names, priority=0, field='')[source]

Marks a method as a pre-transition hook. The hook will be called just before changing a WorkflowEnabled object state, with the same *args and **kwargs as the actual implementation.

xworkflows.transition_check(*names, priority=0, field='')[source]

Marks a method as a transition check hook.

The hook will be called when using is_available() and before running the implementation, without any args, and should return a boolean indicating whether the transition may proceed.

xworkflows.after_transition(*names, priority=0, field='')[source]

Marks a method as a post-transition hook

The hook will be called immediately after the state update, with:

  • res, return value of the actual implementation
  • *args and **kwargs that were passed to the implementation
xworkflows.on_leave_state(*names, priority=0, field='')[source]

Marks a method as a pre-transition hook to call when the object leaves one of the given states.

The hook will be called with the same arguments as a before_transition() hook.

xworkflows.on_enter_state(*names, priority=0, field='')[source]

Marks a method as a post-transition hook to call just after changing the state to one of the given states.

The hook will be called with the same arguments as a after_transition() hook.

Calling hooks

xworkflows.HOOK_BEFORE

The kind of before_transition() hooks

xworkflows.HOOK_CHECK

The kind of transition_check() hooks

xworkflows.HOOK_AFTER

The kind of after_transition() hooks

xworkflows.HOOK_ON_ENTER

The kind of on_leave_state() hooks

xworkflows.HOOK_ON_LEAVE

The kind of on_enter_state() hooks

class base.Hook

Describes a hook, including its kind, priority and the list of transitions it applies to.

kind

One of HOOK_BEFORE, HOOK_AFTER, HOOK_CHECK, HOOK_ON_ENTER or HOOK_ON_LEAVE; the kind of hook.

priority

The priority of the hook, as an integer defaulting to 0. Hooks with higher priority will be executed first; hooks with the same priority will be sorted according to the function name.

Type:int
function

The actual hook function to call. Arguments passed to that function depend on the hook’s kind.

Type:callable
names

Name of states or transitions this hook applies to; will be ('*',) if the hook applies to all states/transitions.

Type:str tuple
applies_to(self, transition[, from_state=None])

Check whether the hook applies to the given Transition and optional source State.

If from_state is None, the test means “could the hook apply to the given transition, in at least one source state”.

If from_state is not None, the test means “does the hook apply to the given transition for this specific source state”.

Returns:bool
__call__(self, *args, **kwargs):

Call the hook

__eq__(self, other)
__ne__(self, other)

Two hooks are “equal” if they wrap the same function, have the same kind, priority and names.

__cmp__(self, other)

Hooks are ordered by descending priority and ascending decorated function name.

Advanced customization

Once WorkflowEnabledMeta has updated the WorkflowEnabled subclass, all transitions – initially defined and automatically added – are replaced with a base.ImplementationProperty instance.

class base.ImplementationProperty

This class holds all objects required to instantiate a ImplementationWrapper whenever the attribute is accessed on an instance.

Internally, it acts as a ‘non-data descriptor’, close to property().

__get__(self, instance, owner)

This method overrides the getattr() behavior:

  • When called without an instance (instance=None), returns itself
  • When called with an instance, this will instantiate a ImplementationWrapper attached to that instance and return it.
add_hook(self, hook)

Register a new Hook.

class base.ImplementationWrapper

This class handles applying a Transition to a WorkflowEnabled object.

instance

The WorkflowEnabled object to modify when calling this wrapper.

field_name

The name of the field modified by this ImplementationProperty (a string)

Type:str
transition

The Transition performed by this object.

Type:Transition
workflow

The Workflow to which this ImplementationProperty relates.

Type:Workflow
implementation

The actual method to call when performing the transition. For undefined implementations, uses noop().

Type:callable
hooks

All hooks that may be applied when performing the related transition.

Type:dict mapping a hook kind to a list of Hook
current_state

Actually a property, retrieve the current state from the instance.

Type:StateWrapper
__call__()

This method allows the TransitionWrapper to act as a function, performing the whole range of checks and hooks before and after calling the actual implementation.

is_available()

Determines whether the wrapped transition implementation can be called. In details:

  • it makes sure that the current state of the instance is compatible with the transition;
  • it calls the transition_check() hooks, if defined.
Return type:bool
base.noop(instance)

The ‘do-nothing’ function called as default implementation of transitions.

Collecting the ImplementationProperty

Warning

This documents private APIs. Use at your own risk.

Building the list of ImplementationProperty for a given WorkflowEnabled, and generating the missing ones, is a complex job.

class base.ImplementationList

This class performs a few low-level operations on a WorkflowEnabled class:

state_field

The name of the attribute (from attr = SomeWorkflow() definition) currently handled.

Type:str
workflow

The Workflow this ImplementationList refers to

implementations

Dict mapping a transition name to the related ImplementationProperty

Type:dict (str => ImplementationProperty)
transitions_at

Dict mapping the name of a transition to the attribute holding its ImplementationProperty:

@transition('foo')
def bar(self):
    pass

will translate into:

self.implementations == {'foo': <ImplementationProperty for 'foo' on 'state': <function bar at 0xdeadbeed>>}
self.transitions_at == {'foo': 'bar'}
custom_implems

Set of name of implementations which were remapped within the workflow.

load_parent_implems(self, parent_implems)

Loads implementations defined in a parent ImplementationList.

Parameters:parent_implems (ImplementationList) – The ImplementationList from a parent
get_custom_implementations(self)

Retrieves definition of custom (non-automatic) implementations from the current list.

Yields:(trname, attr, implem): Tuples containing the transition name, the name of the attribute its implementation is stored at, and that implementation (a ImplementationProperty).
should_collect(self, value)

Whether a given attribute value should be collected in the current list.

Checks that it is a TransitionWrapper, for a Transition of the current Workflow, and relates to the current state_field.

collect(self, attrs)

Collects all TransitionWrapper from an attribute dict if they verify should_collect().

Raises:ValueError If two TransitionWrapper for a same Transition are defined in the attributes.
add_missing_implementations(self)

Registers noop() ImplementationProperty for all Transition that weren’t collected in the collect() step.

register_hooks(self, cls)

Walks the class attributes and collects hooks from those with a xworkflows_hook attribute (through register_function_hooks())

register_function_hooks(self, func)

Retrieves hook definitions from the given function, and registers them on the related ImplementationProperty.

_may_override(self, implem, other)

Checks whether the implem ImplementationProperty is a valid override for the other ImplementationProperty.

Rules are:

fill_attrs(self, attrs)

Adds all ImplementationProperty from implementations to the given attributes dict, unless _may_override() prevents the operation.

transform(self, attrs)
Parameters:attrs (dict) – Mapping holding attribute declarations from a class definition

Performs the following actions, in order: