philiprehberger-state-machine
v0.5.0364 downloadsMIT
Lightweight finite state machine with guards, callbacks, and visualization.
$
pip install philiprehberger-state-machineREADME
philiprehberger-state-machine
Lightweight finite state machine with guards, callbacks, and visualization.
Installation
pip install philiprehberger-state-machine
Usage
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["pending", "confirmed", "shipped", "delivered"],
initial="pending",
transitions=[
("pending", "confirmed", "confirm"),
("confirmed", "shipped", "ship"),
("shipped", "delivered", "deliver"),
],
)
sm.trigger("confirm")
print(sm.state) # "confirmed"
Checking Available Transitions
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["pending", "confirmed", "shipped"],
initial="pending",
transitions=[
("pending", "confirmed", "confirm"),
("confirmed", "shipped", "ship"),
],
)
sm.can("confirm") # True
sm.can("ship") # False
Callbacks
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["pending", "confirmed", "shipped"],
initial="pending",
transitions=[
("pending", "confirmed", "confirm"),
("confirmed", "shipped", "ship"),
],
)
sm.on_enter("confirmed", lambda state, event: print(f"Entered {state} via {event}"))
sm.on_exit("pending", lambda state, event: print(f"Left {state} via {event}"))
sm.trigger("confirm")
# Left pending via confirm
# Entered confirmed via confirm
Guard Conditions
Guards are optional callables that receive a context dict and must return True to allow the transition. If a guard returns falsy, InvalidTransitionError is raised.
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["draft", "published"],
initial="draft",
transitions=[],
)
sm.add_transition("draft", "published", "publish", guard=lambda ctx: ctx.get("has_title", False))
sm.trigger("publish", context={"has_title": True}) # succeeds
print(sm.state) # "published"
Transition Context
Pass a context dict to trigger() to share data with guards and callbacks.
sm.trigger("confirm", context={"user": "alice", "approved": True})
If no context is provided, an empty dict is passed to guards.
Wildcard Transitions
Use "*" as the source state to define a transition that can fire from any state.
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["idle", "running", "paused", "error"],
initial="idle",
transitions=[
("idle", "running", "start"),
("running", "paused", "pause"),
("*", "error", "fail"),
],
)
sm.trigger("start")
print(sm.state) # "running"
sm.trigger("fail")
print(sm.state) # "error"
Wildcard transitions are checked after exact state matches, so a specific transition always takes priority.
Global transition hook
Register a listener that fires after every successful transition. Useful for logging, metrics, or auditing.
import logging
from philiprehberger_state_machine import StateMachine
log = logging.getLogger(__name__)
sm = StateMachine(
states=["pending", "confirmed", "shipped"],
initial="pending",
transitions=[
("pending", "confirmed", "confirm"),
("confirmed", "shipped", "ship"),
],
)
unsubscribe = sm.on_transition(
lambda fr, to, ev, ctx: log.info("transition %s -> %s via %s", fr, to, ev)
)
sm.trigger("confirm")
sm.trigger("ship")
unsubscribe() # remove the listener
The callback receives (from_state, to_state, event, context). Listeners fire in registration order, after on_exit and on_enter callbacks. Listeners do not fire when a guard rejects a transition.
Transition History
Track all past transitions with timestamps using transition_history.
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["pending", "confirmed", "shipped"],
initial="pending",
transitions=[
("pending", "confirmed", "confirm"),
("confirmed", "shipped", "ship"),
],
)
sm.trigger("confirm")
sm.trigger("ship")
for record in sm.transition_history:
print(f"{record.from_state} -> {record.to_state} via {record.event} at {record.timestamp}")
# pending -> confirmed via confirm at 1711929600.123
# confirmed -> shipped via ship at 1711929600.456
History and Reset
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["pending", "confirmed", "shipped"],
initial="pending",
transitions=[
("pending", "confirmed", "confirm"),
("confirmed", "shipped", "ship"),
],
)
sm.trigger("confirm")
sm.trigger("ship")
print(sm.history) # ["pending", "confirmed"]
sm.reset()
print(sm.state) # "pending"
print(sm.history) # []
Timeout-Based Automatic Transitions
Define transitions that fire automatically after a state has been active for a given number of seconds.
from philiprehberger_state_machine import StateMachine
import time
sm = StateMachine(
states=["idle", "processing", "timeout_state"],
initial="idle",
transitions=[("idle", "processing", "start")],
)
sm.add_timeout("processing", "timeout_state", seconds=5.0)
sm.trigger("start")
print(sm.state) # "processing"
time.sleep(6)
print(sm.state) # "timeout_state"
Snapshot and Restore
Capture and restore the machine's state and history for serialization or checkpointing.
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["a", "b", "c"],
initial="a",
transitions=[("a", "b", "go"), ("b", "c", "go")],
)
sm.trigger("go")
snap = sm.snapshot()
print(snap) # {"state": "b", "history": ["a"]}
sm.trigger("go")
print(sm.state) # "c"
sm.restore(snap)
print(sm.state) # "b"
Visualization Export
Export the state machine as a DOT (Graphviz) or Mermaid diagram string.
from philiprehberger_state_machine import StateMachine
sm = StateMachine(
states=["pending", "confirmed", "shipped"],
initial="pending",
transitions=[
("pending", "confirmed", "confirm"),
("confirmed", "shipped", "ship"),
],
)
print(sm.to_dot())
# digraph StateMachine {
# rankdir=LR;
#
# "pending" [shape=doublecircle];
# "confirmed" [shape=circle];
# "shipped" [shape=circle];
#
# "pending" -> "confirmed" [label="confirm"];
# "confirmed" -> "shipped" [label="ship"];
# }
print(sm.to_mermaid())
# stateDiagram-v2
# [*] --> pending
# pending --> confirmed : confirm
# confirmed --> shipped : ship
API
| Function / Class | Description |
|---|---|
StateMachine(states, initial, transitions) | Create a state machine with given states, initial state, and transitions |
StateMachine.state | Current state (read-only property) |
StateMachine.history | List of past states (read-only property) |
StateMachine.transition_history | List of TransitionRecord objects with timestamps (read-only property) |
StateMachine.trigger(event, context=None) | Execute a transition or raise InvalidTransitionError. Pass optional context dict to guards. |
StateMachine.can(event) | Return whether the event is valid from the current state |
StateMachine.add_transition(from_state, to_state, event, guard=None) | Add a transition with an optional guard callable. Use "*" as from_state for wildcard. |
StateMachine.on_enter(state, callback) | Register a callback for entering a state |
StateMachine.on_exit(state, callback) | Register a callback for exiting a state |
StateMachine.on_transition(callback) | Register a global listener fired after every successful transition with (from_state, to_state, event, context). Returns an unsubscribe closure. |
StateMachine.remove_transition_listener(callback) | Remove a previously registered transition listener. Returns True if removed. |
StateMachine.reset() | Reset to initial state and clear history |
StateMachine.add_timeout(state, target, seconds) | Define an automatic transition after seconds in state |
StateMachine.snapshot() | Return a serializable dict of current state and history |
StateMachine.restore(snapshot) | Restore the machine from a snapshot dict |
StateMachine.to_dot() | Return a DOT/Graphviz string of the state machine |
StateMachine.to_mermaid() | Return a Mermaid state diagram string |
TransitionRecord | Frozen dataclass with from_state, to_state, event, and timestamp fields |
InvalidTransitionError | Raised on invalid transitions; has .state and .event attributes |
Development
pip install -e .
python -m pytest tests/ -v
Support
If you find this project useful: