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:
- During class definition, the state manager returns the managed state. All methods on the state manager recognise these managed states and handle them appropriately.
- 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.
- 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 toTrue
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:
- Original and final states can be specified, prohibiting arbitrary state changes.
- The transition method can do additional validation and housekeeping.
- Combined with the
with_roles()
decorator andRoleMixin
, transitions provide access control for state changes. - 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 toclass_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
- states –
ManagedState
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. Useadd_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. Useadd_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()
, usehelp(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 aStateManager
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.
-
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 sourceLabeledEnum
.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
orManagedStateGroup
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 byStateTransition
.-
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