A lightweight, object-oriented Python state machine implementation with many extensions
The core state machine functionality provides the foundation for all state machine operations in the transitions library. The Machine class manages states, transitions, models, and events with extensive configuration options and callback support.
The central state machine implementation that manages states, transitions, and models.
class Machine:
def __init__(
self,
model='self',
states=None,
initial='initial',
transitions=None,
send_event=False,
auto_transitions=True,
ordered_transitions=False,
ignore_invalid_triggers=None,
before_state_change=None,
after_state_change=None,
name=None,
queued=False,
prepare_event=None,
finalize_event=None,
model_attribute='state',
model_override=False,
on_exception=None,
on_final=None,
**kwargs
):
"""
Initialize a state machine.
Parameters:
- model: The object(s) whose states we want to manage. 'self' uses the machine itself.
- states: List of state names or State objects
- initial: Name of the initial state
- transitions: List of transition dictionaries
- send_event: When True, arguments are wrapped in EventData
- auto_transitions: When True, every state gets to_{state}() convenience methods
- ordered_transitions: When True, add ordered transitions between consecutive states
- ignore_invalid_triggers: When True, invalid triggers are silently ignored
- before_state_change: Callbacks executed before any transition
- after_state_change: Callbacks executed after any transition
- name: Name of the machine instance for logging
- queued: When True, transitions are queued and executed sequentially
- prepare_event: Callbacks executed when an event is triggered
- finalize_event: Callbacks executed after all transition callbacks
- model_attribute: Name of the model attribute that stores the state
- model_override: When True, override model methods with trigger methods
- on_exception: Callbacks executed when Event raises an exception
- on_final: Callbacks executed when a final state is reached
"""Methods for managing states in the machine.
def add_state(self, states, on_enter=None, on_exit=None, ignore_invalid_triggers=None, **kwargs):
"""
Add new state(s) to the machine.
Parameters:
- states: State name, State object, or list of either
- on_enter: Callbacks triggered when state is entered
- on_exit: Callbacks triggered when state is exited
- ignore_invalid_triggers: Override machine's ignore_invalid_triggers setting
"""
def add_states(self, states, on_enter=None, on_exit=None, ignore_invalid_triggers=None, **kwargs):
"""Alias for add_state that handles multiple states."""
def get_state(self, state):
"""
Return the State instance with the passed name.
Parameters:
- state: Name of the state to retrieve
Returns:
State object
"""
def is_state(self, state, model):
"""
Check whether the current state matches the named state.
Parameters:
- state: Name of the state to check
- model: Model to check state for
Returns:
bool: True if current state matches
"""
def get_model_state(self, model):
"""
Get the state of a specific model.
Parameters:
- model: Model to get state for
Returns:
Current state of the model
"""
def set_state(self, state, model=None):
"""
Set the current state.
Parameters:
- state: Name of the state to set
- model: Model to set state for (all models if None)
"""Methods for managing transitions between states.
def add_transition(
self,
trigger,
source,
dest,
conditions=None,
unless=None,
before=None,
after=None,
prepare=None,
**kwargs
):
"""
Create a new transition.
Parameters:
- trigger: Name of the triggering method
- source: Source state name or list of source states
- dest: Destination state name
- conditions: Callbacks that must return True for transition to execute
- unless: Callbacks that must return False for transition to execute
- before: Callbacks executed before the transition
- after: Callbacks executed after the transition
- prepare: Callbacks executed before condition checks
"""
def add_transitions(self, transitions):
"""
Add several transitions at once.
Parameters:
- transitions: List of transition dictionaries
"""
def add_ordered_transitions(
self,
states=None,
trigger='next_state',
loop=True,
loop_includes_initial=True,
conditions=None,
unless=None,
before=None,
after=None,
prepare=None,
**kwargs
):
"""
Add transitions that move linearly from state to state.
Parameters:
- states: List of state names (uses machine states if None)
- trigger: Name of the trigger method
- loop: Whether to loop back to first state from last state
- loop_includes_initial: Whether the loop includes the initial state
- conditions: Conditions for all transitions
- unless: Unless conditions for all transitions
- before: Before callbacks for all transitions
- after: After callbacks for all transitions
- prepare: Prepare callbacks for all transitions
"""
def get_transitions(self, trigger="", source="*", dest="*"):
"""
Return transitions matching the criteria.
Parameters:
- trigger: Trigger name to filter by (empty string for all)
- source: Source state to filter by ("*" for all)
- dest: Destination state to filter by ("*" for all)
Returns:
List of matching Transition objects
"""
def remove_transition(self, trigger, source="*", dest="*"):
"""
Remove transitions matching the criteria.
Parameters:
- trigger: Trigger name
- source: Source state ("*" for all)
- dest: Destination state ("*" for all)
"""Methods for managing models attached to the machine.
def add_model(self, model, initial=None):
"""
Register a model with the state machine.
Parameters:
- model: Model object to register
- initial: Initial state for this model (uses machine initial if None)
"""
def remove_model(self, model):
"""
Remove a model from the state machine.
Parameters:
- model: Model object to remove
"""Methods for triggering events and managing callbacks.
def dispatch(self, trigger, *args, **kwargs):
"""
Trigger an event on all models assigned to the machine.
Parameters:
- trigger: Name of the trigger method
- args: Positional arguments passed to callbacks
- kwargs: Keyword arguments passed to callbacks
Returns:
bool: True if at least one transition was successful
"""
def get_triggers(self, *args):
"""
Collect all triggers available from certain states.
Parameters:
- args: State names to get triggers for (all states if empty)
Returns:
List of trigger names
"""Helper methods for callback handling and state management.
def callbacks(self, funcs, event_data):
"""
Trigger a list of callbacks with event data.
Parameters:
- funcs: List of callback functions
- event_data: EventData object to pass to callbacks
"""
def callback(self, func, event_data):
"""
Trigger a single callback function.
Parameters:
- func: Callback function or method name
- event_data: EventData object to pass to callback
"""
def resolve_callable(self, func, event_data):
"""
Convert a model's property/method name or path to callable.
Parameters:
- func: Function, method name, or dotted path
- event_data: EventData object for context
Returns:
Callable function
"""Machine properties for state and configuration access.
@property
def initial(self):
"""Get or set the initial state name."""
@property
def has_queue(self):
"""Return boolean indicating if machine has queue enabled."""
@property
def model(self):
"""List of models attached to the machine."""
@property
def before_state_change(self):
"""Callbacks executed before any state transition."""
@property
def after_state_change(self):
"""Callbacks executed after any state transition."""
@property
def prepare_event(self):
"""Callbacks executed when an event is triggered."""
@property
def finalize_event(self):
"""Callbacks executed after all transition callbacks."""
@property
def on_exception(self):
"""Callbacks executed when an Event raises an exception."""
@property
def on_final(self):
"""Callbacks executed when a final state is reached."""def listify(obj):
"""
Wrap a passed object into a list if it's not already a list or tuple.
Parameters:
- obj: Object to convert to list (returns empty list if None)
Returns:
list: List containing the object, or empty list if obj is None
"""class MachineError(Exception):
"""
Exception used for issues related to state transitions and current states.
Parameters:
- value: Error message or description
"""
def __init__(self, value): ...from transitions import Machine
# Define a simple model
class TrafficLight:
pass
# Define states and transitions
states = ['red', 'yellow', 'green']
transitions = [
{'trigger': 'go', 'source': 'red', 'dest': 'green'},
{'trigger': 'slow_down', 'source': 'green', 'dest': 'yellow'},
{'trigger': 'stop', 'source': 'yellow', 'dest': 'red'}
]
# Create the machine
light = TrafficLight()
machine = Machine(model=light, states=states, transitions=transitions, initial='red')
# Use the state machine
print(light.state) # 'red'
light.go()
print(light.state) # 'green'from transitions import Machine
class Robot:
def __init__(self):
self.battery = 100
def check_battery(self):
print(f"Battery level: {self.battery}%")
return self.battery > 20
def consume_battery(self):
self.battery -= 10
print(f"Battery consumed, remaining: {self.battery}%")
states = ['idle', 'working', 'charging']
transitions = [
{
'trigger': 'start_work',
'source': 'idle',
'dest': 'working',
'conditions': 'check_battery',
'after': 'consume_battery'
},
{'trigger': 'finish_work', 'source': 'working', 'dest': 'idle'},
{'trigger': 'charge', 'source': ['idle', 'working'], 'dest': 'charging'}
]
robot = Robot()
machine = Machine(model=robot, states=states, transitions=transitions, initial='idle')
robot.start_work() # Checks battery and consumes itInstall with Tessl CLI
npx tessl i tessl/pypi-transitions