States and transitions

StateManager wraps a SQLAlchemy column with a LabeledEnum to facilitate state inspection, and to control state change via transitions. Sample usage:

class MY_STATE(LabeledEnum):
    DRAFT = (0, "Draft")
    PENDING = (1, 'pending', "Pending")
    PUBLISHED = (2, "Published")

    UNPUBLISHED = {DRAFT, PENDING}


# Classes can have more than one state variable
class REVIEW_STATE(LabeledEnum):
    UNSUBMITTED = (0, "Unsubmitted")
    PENDING = (1, "Pending")
    REVIEWED = (2, "Reviewed")


class MyPost(BaseMixin, db.Model):
    __tablename__ = 'my_post'

    # The underlying state value columns
    # (more than one state variable can exist)
    _state = db.Column('state', db.Integer,
        StateManager.check_constraint('state', MY_STATE),
        default=MY_STATE.DRAFT, nullable=False)
    _reviewstate = db.Column('reviewstate', db.Integer,
        StateManager.check_constraint('state', REVIEW_STATE),
        default=REVIEW_STATE.UNSUBMITTED, nullable=False)

    # The state managers controlling the columns
    state = StateManager('_state', MY_STATE, doc="The post's state")
    reviewstate = StateManager('_reviewstate', REVIEW_STATE,
        doc="Reviewer's state")

    # Datetime for the additional states and transitions
    datetime = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

    # Additional states:

    # RECENT = PUBLISHED + in the last one hour
    state.add_conditional_state('RECENT', state.PUBLISHED,
        lambda post: post.datetime > datetime.utcnow() - timedelta(hours=1))

    # REDRAFTABLE = DRAFT or PENDING or RECENT
    state.add_state_group('REDRAFTABLE',
        state.DRAFT, state.PENDING, state.RECENT)

    # Transitions change FROM one state TO another, and can have
    # an additional if_ condition (a callable) that must return True
    @state.transition(state.DRAFT, state.PENDING, if_=reviewstate.UNSUBMITTED)
    def submit(self):
        pass

    # Transitions can coordinate across state managers. All of them
    # must be in a valid FROM state for the transition to be available.
    # Transitions can also specify arbitrary metadata such as this `title`
    # attribute (on any of the decorators). These are made available in a
    # `data` dictionary, accessible here as `publish.data`
    @state.transition(state.UNPUBLISHED, state.PUBLISHED, title="Publish")
    @reviewstate.transition(reviewstate.UNSUBMITTED, reviewstate.PENDING)
    def publish(self):
        # A transition can do additional housekeeping
        self.datetime = datetime.utcnow()

    # A transition can use a conditional state. The condition is evaluated
    # before the transition can proceed
    @state.transition(state.RECENT, state.PENDING)
    @reviewstate.transition(reviewstate.PENDING, reviewstate.UNSUBMITTED)
    def undo(self):
        pass

    # Transitions can be defined FROM a group of states, but the TO
    # state must always be an individual state
    @state.transition(state.REDRAFTABLE, state.DRAFT)
    def redraft(self):
        pass

    # Transitions can abort without changing state, with or without raising
    # an exception to the caller
    @state.transition(state.REDRAFTABLE, state.DRAFT)
    def faulty_transition_examples(self):
        # Cancel the transition, but don't raise an exception to the caller
        raise AbortTransition()
        # Cancel the transition and return a result to the caller
        raise AbortTransition('failed')
        # Need to return a data structure? That works as well
        raise AbortTransition((False, 'faulty_failure'))
        raise AbortTransition({'status': 'error', 'error': 'faulty_failure'})
        # If any other exception is raised, it is passed up to the caller
        raise ValueError("Faulty transition")

    # The requires decorator specifies a transition that does not change
    # state. It can be used to limit a method's availability
    @state.requires(state.PUBLISHED)
    def send_email_alert(self):
        pass

Defining states and transitions

Adding a StateManager to the class links the underlying column (specified as a string) to the LabeledEnum (specified as an object). The StateManager is read-only and state can only be mutated via transitions. The LabeledEnum is not required after this point. All symbol names in it are available as attributes on the state manager henceforth (as instances of ManagedState).

Conditional states can be defined with add_conditional_state() as a combination of an existing state and a validator that receives the object (the instance of the class the StateManager is present on). This can be used to evaluate for additional conditions. For example, to distinguish between a static “published” state and a dynamic “recently published” state. add_conditional_state() also takes an optional class_validator parameter that is used for queries against the class (see below for query examples).

State groups can be defined with add_state_group(). These are similar to grouped values in a LabeledEnum, but can also contain conditional states, and are stored as instances of ManagedStateGroup. Grouped values in a LabeledEnum are more efficient for testing state against, so those should be preferred if the group does not contain a conditional state.

Transitions connect one managed state or group to another state (but not group). Transitions are defined as methods and decorated with transition(), which transforms them into instances of StateTransition, a callable class. If the transition raises an exception, the state change is aborted. Transitions may also abort without changing state using AbortTransition. Transitions have two additional attributes, is_available, a boolean property which indicates if the transition is currently available, and data, a dictionary that contains all additional parameters passed to the transition() decorator.

Transitions can be chained to coordinate a state change across state managers if the class has more than one. All state managers must be in a valid from state for the transition to be available. A dictionary of currently available transitions can be obtained from the state manager using the transitions() method.

Queries

The current state of the object can be retrieved by calling the state attribute or reading its value attribute:

post = MyPost(_state=MY_STATE.DRAFT)
post.state() == MY_STATE.DRAFT
post.state.value == MY_STATE.DRAFT

The label associated with the state value can be accessed from the label attribute:

post.state.label == "Draft"          # This is the string label from MY_STATE.DRAFT
post.submit()                        # Change state from DRAFT to PENDING
post.state.label.name == 'pending'   # Is the NameTitle tuple from MY_STATE.PENDING
post.state.label.title == "Pending"  # The title part of NameTitle

States can be tested by direct reference using the names they were originally defined with in the LabeledEnum:

post.state.DRAFT        # True
post.state.is_draft     # True (is_* attrs are lowercased aliases to states)
post.state.PENDING      # False (since it's a draft)
post.state.UNPUBLISHED  # True (grouped state values work as expected)
post.publish()          # Change state from DRAFT to PUBLISHED
post.state.RECENT       # True (calls the validator if the base state matches)

States can also be used for database queries when accessed from the class:

# Generates MyPost._state == MY_STATE.DRAFT
MyPost.query.filter(MyPost.state.DRAFT)

# Generates MyPost._state.in_(MY_STATE.UNPUBLISHED)
MyPost.query.filter(MyPost.state.UNPUBLISHED)

# Generates and_(MyPost._state == MY_STATE.PUBLISHED,
#     MyPost.datetime > datetime.utcnow() - timedelta(hours=1))
MyPost.query.filter(MyPost.state.RECENT)

This works because StateManager, ManagedState and ManagedStateGroup behave in three different ways, depending on context:

  1. During class definition, the state manager returns the managed state. All methods on the state manager recognise these managed states and handle them appropriately.
  2. After class definition, the state manager returns the result of calling the managed state instance. If accessed via the class, the managed state returns a SQLAlchemy filter condition.
  3. After class definition, if accessed via an instance, the managed state returns itself wrapped in ManagedStateWrapper (which holds context for the instance). This is an object that evaluates to True if the state is active, False otherwise. It also provides pass-through access to all attributes of the managed state.

States can be changed via transitions, defined as methods with the transition() decorator. They add more power and safeguards over direct state value changes:

  1. Original and final states can be specified, prohibiting arbitrary state changes.
  2. The transition method can do additional validation and housekeeping.
  3. Combined with the with_roles() decorator and RoleMixin, transitions provide access control for state changes.
  4. Signals are raised before and after a successful transition, or in case of failures, allowing for the attempts to be logged.
class coaster.sqlalchemy.statemanager.StateManager(propname, lenum, doc=None)[source]

Wraps a property with a LabeledEnum to facilitate state inspection and control state changes.

This is the main export of this module.

Parameters:
  • propname (str) – Name of the property that is to be wrapped
  • lenum (LabeledEnum) – The LabeledEnum containing valid values
  • doc (str) – Optional docstring
add_conditional_state(name, state, validator, class_validator=None, cache_for=None, label=None)[source]

Add a conditional state that combines an existing state with a validator that must also pass. The validator receives the object on which the property is present as a parameter.

Parameters:
  • name (str) – Name of the new state
  • state (ManagedState) – Existing state that this is based on
  • validator – Function that will be called with the host object as a parameter
  • class_validator – Function that will be called when the state is queried on the class instead of the instance. Falls back to validator if not specified. Receives the class as the parameter
  • cache_for – Integer or function that indicates how long validator’s result can be cached (not applicable to class_validator). None implies no cache, 0 implies indefinite cache (until invalidated by a transition) and any other integer is the number of seconds for which to cache the assertion
  • label – Label for this state (string or 2-tuple)

TODO: cache_for’s implementation is currently pending a test case demonstrating how it will be used.

add_state_group(name, *states)[source]

Add a group of managed states. Groups can be specified directly in the LabeledEnum. This method is only useful for grouping a conditional state with existing states. It cannot be used to form a group of groups.

Parameters:
  • name (str) – Name of this group
  • statesManagedState instances to be grouped together
static check_constraint(column, lenum, **kwargs)[source]

Returns a SQL CHECK constraint string given a column name and a LabeledEnum.

Alembic may not detect the CHECK constraint when autogenerating migrations, so you may need to do this manually using the Python console to extract the SQL string:

from coaster.sqlalchemy import StateManager
from your_app.models import YOUR_ENUM

print str(StateManager.check_constraint('your_column', YOUR_ENUM).sqltext)
Parameters:
  • column (str) – Column name
  • lenum (LabeledEnum) – LabeledEnum to retrieve valid values from
  • kwargs – Additional options passed to CheckConstraint
requires(from_, if_=None, **data)[source]

Decorates a method that may be called if the given state is currently active. Registers a transition internally, but does not change the state.

Parameters:
  • from – Required state to allow this call (can be a state group)
  • if – Validator(s) that, given the object, must all return True for the call to proceed
  • data – Additional metadata, stored on the StateTransition object as a data attribute
transition(from_, to, if_=None, **data)[source]

Decorates a method to transition from one state to another. The decorated method can accept any necessary parameters and perform additional processing, or raise an exception to abort the transition. If it returns without an error, the state value is updated automatically. Transitions may also abort without raising an exception using AbortTransition.

Parameters:
  • from – Required state to allow this transition (can be a state group)
  • to – The state of the object after this transition (automatically set if no exception is raised)
  • if – Validator(s) that, given the object, must all return True for the transition to proceed
  • data – Additional metadata, stored on the StateTransition object as a data attribute
class coaster.sqlalchemy.statemanager.ManagedState(name, statemanager, value, label=None, validator=None, class_validator=None, cache_for=None)[source]

Represents a state managed by a StateManager. Do not use this class directly. Use add_conditional_state() instead.

is_conditional

This is a conditional state

is_direct

This is a direct state (scalar state without a condition)

is_scalar

This is a scalar state (not a group of states, and may or may not have a condition)

class coaster.sqlalchemy.statemanager.ManagedStateGroup(name, statemanager, states)[source]

Represents a group of managed states in a StateManager. Do not use this class directly. Use add_state_group() instead.

class coaster.sqlalchemy.statemanager.StateTransition(func, statemanager, from_, to, if_=None, data=None)[source]

Helper for transitions from one state to another. Do not use this class directly. Use the StateManager.transition() decorator instead, which creates instances of this class.

To access the decorated function with help(), use help(obj.func).

class coaster.sqlalchemy.statemanager.StateManagerWrapper(statemanager, obj: Optional[T], cls: Optional[Type[T]])[source]

Wraps StateManager with the context of the containing object. Automatically constructed when a StateManager is accessed from either a class or an instance.

bestmatch()[source]

Best matching current scalar state (direct or conditional), only applicable when accessed via an instance.

current()[source]

All states and state groups that are currently active.

group(items, keep_empty=False)[source]

Given an iterable of instances, groups them by state using ManagedState instances as dictionary keys. Returns a dict that preserves the order of states from the source LabeledEnum.

Parameters:keep_empty (bool) – If True, empty states are included in the result
label

Label for the current state’s value (using bestmatch()).

transitions(current=True)[source]

Returns available transitions for the current state, as a dictionary of name: StateTransitionWrapper.

Parameters:current (bool) – Limit to transitions available in obj. current_access()
transitions_for(roles=None, actor=None, anchors=())[source]

For use on RoleMixin classes: returns currently available transitions for the specified roles or actor as a dictionary of name: StateTransitionWrapper.

value

The current state value.

class coaster.sqlalchemy.statemanager.ManagedStateWrapper(mstate, obj, cls=None)[source]

Wraps a ManagedState or ManagedStateGroup with an object or class, and otherwise provides transparent access to contents.

This class is automatically constructed by StateManager.

class coaster.sqlalchemy.statemanager.StateTransitionWrapper(statetransition, obj)[source]

Wraps StateTransition with the context of the object it is accessed from. Automatically constructed by StateTransition.

data

Dictionary containing all additional parameters to the transition() decorator.

is_available

Property that indicates whether this transition is currently available.

exception coaster.sqlalchemy.statemanager.StateTransitionError(description: Optional[str] = None, response: Optional[Response] = None)[source]

Raised if a transition is attempted from a non-matching state

exception coaster.sqlalchemy.statemanager.AbortTransition(result=None)[source]

Transitions may raise AbortTransition to return without changing state. The parameter to this exception is returned as the transition’s result.

This exception is a signal to StateTransition and will not be raised to the transition’s caller.

Parameters:result – Value to return to the transition’s caller
coaster.sqlalchemy.statemanager.transition_error = <blinker.base.NamedSignal object at 0x7f6999701d50; 'transition-error'>

Signal raised when a transition fails validation

coaster.sqlalchemy.statemanager.transition_before = <blinker.base.NamedSignal object at 0x7f6999701e10; 'transition-before'>

Signal raised before a transition (after validation)

coaster.sqlalchemy.statemanager.transition_after = <blinker.base.NamedSignal object at 0x7f6999701e50; 'transition-after'>

Signal raised after a successful transition

coaster.sqlalchemy.statemanager.transition_exception = <blinker.base.NamedSignal object at 0x7f6999701e90; 'transition-exception'>

Signal raised when a transition raises an exception