Skip to main content

Lightweight finite state machine with guards and callbacks.

Project description

philiprehberger-state-machine

Tests PyPI version Last updated

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 project useful:

Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

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.1.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.

File details

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

File metadata

File hashes

Hashes for philiprehberger_state_machine-0.3.1.tar.gz
Algorithm Hash digest
SHA256 ceacb962acfacf9da3657ce7fabfaf5a5677e91a0943622a648bbcd54c89ef0b
MD5 587aec7b09b8f1cc397d9bfd243b57ba
BLAKE2b-256 1e80d36fb614b03b426bd55d522dbf3ad372b9b8493c228fad8f9e5b83e56f84

See more details on using hashes here.

File details

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

File metadata

File hashes

Hashes for philiprehberger_state_machine-0.3.1-py3-none-any.whl
Algorithm Hash digest
SHA256 9051403fdddd91a56fae674c156563767ae478273d157c0ddeb1fbc106e6fb4c
MD5 dd14c57473aa1ae3a241633d52315b17
BLAKE2b-256 ea44d72a17a0b827cda741e3dc71ab46b721f10136090c55d65c275611e51506

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