Skip to content

Engine

Instance-level introspection over a class's State field. Everything here delegates to StateDescriptor — no state machine logic lives in this module.

>>> from open_fsm import (
...     CURRENT,
...     StateEngine,
...     get_available_transitions,
...     get_outgoing_transitions,
...     get_state_field,
...     get_transitions,
... )

CURRENT

The default state argument, meaning "whatever state the flow is in now". A marker rather than None, because None — like 0 or '' — is a legitimate state, and it is compared by identity so it can never collide with one.

>>> article = publication.Article('No body yet')
>>> get_outgoing_transitions(article, CURRENT) == article.get_outgoing_transitions()
True

get_state_field(owner)

The State declared by owner or one of its bases.
Returns The declared State
Raises TypeError if the class declares no State, or more than one

The lookup walks reversed(owner.__mro__) and caches the result on owner under _fsm_state_field_cache. The cache is read from owner.__dict__, not inherited, so a subclass redeclaring the field recomputes it.

>>> get_state_field(publication.Article).propname
'__fsm_state'

get_transitions(flow)

Every transition declared by the flow's class, keyed by transition method. Ignores the current state — this is the whole machine.

>>> sorted(method.slug for method in get_transitions(article))
['approve', 'archive', 'publish', 'reject', 'submit']

get_outgoing_transitions(flow, state=CURRENT)

Transitions leaving state, conditions not evaluated.

>>> [transition.slug for transition in get_outgoing_transitions(article)]
['archive', 'submit']

Pass a state to ask about one the flow is not in. The flow is not modified:

>>> sorted(
...     transition.slug
...     for transition in get_outgoing_transitions(article, publication.ReviewState.APPROVED)
... )
['archive', 'publish']

get_available_transitions(flow, state=CURRENT)

The outgoing transitions whose conditions pass for flow.

>>> [transition.slug for transition in get_available_transitions(article)]
['archive']

Conditions are evaluated against the flow as it is now

Even when state is one the flow is not in. The answer is "what could this instance do from there", not "what will be possible once it gets there".

StateEngine

A mixin exposing the three functions above on the instance, without repeating it or its state.

Method Equivalent to
get_transitions() get_transitions(self)
get_outgoing_transitions() get_outgoing_transitions(self, CURRENT)
get_available_transitions() get_available_transitions(self, CURRENT)

It declares no __init__, no metaclass and no fields of its own — only a ClassVar annotation for the field cache, which does not exist as an attribute until the first lookup. That is what makes it safe to mix into an ORM model or a dataclass.

The methods take no arguments, deliberately, so the two questions are never confused:

>>> article.get_outgoing_transitions(publication.ReviewState.APPROVED)
Traceback (most recent call last):
    ...
TypeError: StateEngine.get_outgoing_transitions() takes 1 positional argument but 2 were given

Use the module-level functions for a state the flow is not in. They also work on a class that cannot inherit the mixin:

>>> class Plain:
...     state = State(['a', 'b'], default='a')
...     @state.transition(source='a', target='b')
...     def go(self):
...         """No mixin."""
>>> [transition.slug for transition in get_outgoing_transitions(Plain())]
['go']