States and transitions

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

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


# 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))

    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):

    # 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 ``
    @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):

    # 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):

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 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.


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 == 'pending'   # Ths 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 (this one 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

# Generates MyPost._state.in_(MY_STATE.UNPUBLISHED)

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

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.

  • 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.

  • 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.

  • 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)
  • 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.

  • 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
  • metadata – Additional metadata, stored on the StateTransition object
transition(from_, to, if_=None, **data)[source]

Decorates a function to transition from one state to another. The decorated function 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.

  • 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
  • metadata – Additional metadata, stored on the StateTransition object
exception coaster.sqlalchemy.statemanager.StateTransitionError(description=None, response=None)[source]

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