Skip to main content

Lightweight finite state machine with guards and callbacks.

Project description

philiprehberger-state-machine

Tests PyPI version GitHub release Last updated License Bug Reports Feature Requests Sponsor

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.

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.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
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.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
InvalidTransitionError Raised on invalid transitions; has .state and .event attributes

Development

pip install -e .
python -m pytest tests/ -v

Support

If you find this package useful, consider giving it a star on GitHub — it helps motivate continued maintenance and development.

LinkedIn More packages

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

philiprehberger_state_machine-0.3.0.tar.gz (8.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

philiprehberger_state_machine-0.3.0-py3-none-any.whl (7.7 kB view details)

Uploaded Python 3

File details

Details for the file philiprehberger_state_machine-0.3.0.tar.gz.

File metadata

File hashes

Hashes for philiprehberger_state_machine-0.3.0.tar.gz
Algorithm Hash digest
SHA256 be029a2b2095c68818bb0dc5bbce866d5242a16854b079c3c04943d22ef190d0
MD5 da02db71b9577ec9f8417a8ae55dfa64
BLAKE2b-256 5c0bbf070a2d46cbcf794f442936e064d409f2a63eb980af10831de5e24ad58f

See more details on using hashes here.

File details

Details for the file philiprehberger_state_machine-0.3.0-py3-none-any.whl.

File metadata

File hashes

Hashes for philiprehberger_state_machine-0.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 cf6504305f6ebe85c3d1cbf687106de352707e7745b0ffa1fa74e30581ef60ee
MD5 9f57c4fc62a93258fb84b90fb7cf5008
BLAKE2b-256 1df08757c27255a018183cd20d600a2b1b5955725d41a447d9305830c6b7d5a9

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page