tessl/pypi-python-statemachine
Python finite state machine library with declarative API for sync and async applications
—
Pending
core-statemachine.mddocs/
Core State Machine and States
Fundamental state machine functionality that provides the foundation for finite state machine implementation in Python. This includes the main StateMachine class, State objects, and basic state management operations.
Capabilities
StateMachine Class
The main class that orchestrates states, events, and transitions with support for both synchronous and asynchronous execution.
class StateMachine:
"""
Main state machine class providing finite state machine functionality.
Parameters:
- model: External object to store state (optional)
- state_field: Model field name for state storage (default: "state")
- start_value: Initial state value if no current state assigned (optional)
- rtc: Run-to-completion model control (default: True)
- allow_event_without_transition: Allow events without transitions (default: False)
- listeners: Optional list of listener objects for callbacks (optional)
"""
def __init__(self, model=None, state_field: str = "state", start_value=None, rtc: bool = True, allow_event_without_transition: bool = False, listeners=None): ...
def send(self, event: str, *args, **kwargs):
"""
Send an event to the state machine to trigger transitions.
Parameters:
- event: Event identifier string
- args: Positional arguments passed to callbacks
- kwargs: Keyword arguments passed to callbacks
Returns:
Result from event processing or callback return values
"""
@property
def current_state(self) -> State:
"""Get the currently active state."""
@property
def current_state_value(self):
"""Get or set the current state value (for external model integration)."""
@current_state_value.setter
def current_state_value(self, value):
"""Set the current state value."""
@property
def states(self) -> States:
"""Get States collection containing all states in the machine."""
@property
def events(self) -> list:
"""Get list of all events in the machine."""
@property
def allowed_events(self) -> list:
"""Get list of events allowed from current state."""
def activate_initial_state(self):
"""Manually activate the initial state (useful for async state machines)."""
def bind_events_to(self, *targets):
"""
Bind event methods to target objects.
Parameters:
- targets: Objects to bind event methods to
"""
def add_listener(self, *listeners):
"""
Add listener objects for callback notifications.
Parameters:
- listeners: Objects that will receive callback notifications
"""
def _graph(self):
"""Get DotGraphMachine instance for diagram generation."""
def _repr_html_(self):
"""HTML representation for Jupyter notebooks."""
def _repr_svg_(self):
"""SVG representation for Jupyter notebooks."""State Class
Represents individual states in the state machine with support for initial/final states, entry/exit actions, and transition building.
class State:
"""
Represents a state in the state machine.
Parameters:
- name: Human-readable state name (derived from variable name if not provided)
- value: Specific value for state storage and retrieval (optional)
- initial: Set True if this is the initial state (default: False)
- final: Set True if this is a final state (default: False)
- enter: Entry action callbacks (optional)
- exit: Exit action callbacks (optional)
"""
def __init__(self, name: str = "", value=None, initial: bool = False, final: bool = False, enter=None, exit=None): ...
def to(self, *states, **kwargs) -> TransitionList:
"""
Create transitions from this state to target states.
Parameters:
- states: Target State objects
- kwargs: Transition parameters (event, cond, unless, on, before, after, etc.)
Returns:
TransitionList containing the created transitions
"""
def from_(self, *states, **kwargs) -> TransitionList:
"""
Create transitions from source states to this state.
Parameters:
- states: Source State objects
- kwargs: Transition parameters
Returns:
TransitionList containing the created transitions
"""
@property
def is_active(self) -> bool:
"""Check if this state is currently active."""
@property
def transitions(self) -> TransitionList:
"""Get transitions originating from this state."""
@property
def id(self) -> str:
"""Get the state identifier."""
@property
def name(self) -> str:
"""Get the human-readable state name."""
@property
def value(self):
"""Get the state value used for storage."""
@property
def initial(self) -> bool:
"""Check if this is the initial state."""
@property
def final(self) -> bool:
"""Check if this is a final state."""
@property
def enter(self):
"""Entry callbacks grouper for adding entry actions."""
@property
def exit(self):
"""Exit callbacks grouper for adding exit actions."""States Collection Class
Helper class for managing collections of State objects, particularly useful when creating states from external sources like Enums.
class States:
"""
Collection class for managing multiple State objects.
Supports dictionary-like access and creation from Enum classes.
"""
def __init__(self, states=None): ...
@classmethod
def from_enum(cls, enum_type, initial, final=None, use_enum_instance=False):
"""
Create States collection from an Enum class.
Parameters:
- enum_type: Enum class to convert to states
- initial: Enum value to mark as initial state
- final: Enum value(s) to mark as final states (optional)
- use_enum_instance: Whether to use enum instances as state values (default: False)
Returns:
States instance with states created from enum values
"""
def append(self, state: State):
"""Add a state to the collection."""
def items(self):
"""Get (id, state) pairs for all states."""
def __getattr__(self, name: str) -> State:
"""Access states by attribute name."""
def __getitem__(self, key: str) -> State:
"""Access states by key name."""
def __iter__(self):
"""Iterate over states."""
def __len__(self) -> int:
"""Get number of states in collection."""Specialized State Classes
Extended state functionality for specific use cases.
class InstanceState(State):
"""Runtime state instance with machine binding."""
class AnyState(State):
"""Special state representing any state for transition building."""Usage Examples
Basic State Machine Definition
from statemachine import StateMachine, State
class DocumentWorkflow(StateMachine):
# Define states
draft = State(initial=True)
review = State()
approved = State()
published = State(final=True)
rejected = State(final=True)
# Define transitions
submit = draft.to(review)
approve = review.to(approved)
publish = approved.to(published)
reject = review.to(rejected) | approved.to(rejected)
revise = review.to(draft) | approved.to(draft)
# Create instance
workflow = DocumentWorkflow()
print(workflow.current_state.id) # "draft"
print(workflow.draft.is_active) # TrueUsing External Model
class Document:
def __init__(self):
self.status = "draft"
# Bind to external model
doc = Document()
workflow = DocumentWorkflow(doc, state_field="status")
# State is stored in document object
workflow.send("submit")
print(doc.status) # "review"Creating States from Enum
from enum import Enum
from statemachine.states import States
class Status(Enum):
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
class JobMachine(StateMachine):
states = States.from_enum(
Status,
initial=Status.PENDING,
final=[Status.COMPLETED, Status.FAILED]
)
start = states.pending.to(states.processing)
complete = states.processing.to(states.completed)
fail = states.processing.to(states.failed)Async State Machine
import asyncio
from statemachine import StateMachine, State
class AsyncMachine(StateMachine):
idle = State(initial=True)
working = State()
done = State(final=True)
start = idle.to(working)
finish = working.to(done)
async def on_enter_working(self):
print("Starting async work...")
await asyncio.sleep(1)
print("Work completed")
# Usage
async def main():
machine = AsyncMachine()
await machine.send_async("start")
await machine.send_async("finish")
asyncio.run(main())Install with Tessl CLI
npx tessl i tessl/pypi-python-statemachine