Skip to main content

Python Finite State Machines made easy.

Project description

Python StateMachine

pypi downloads total downloads Coverage report Documentation Status GitHub commits since last release (main)

Expressive statecharts and FSMs for modern Python.

Welcome to python-statemachine, an intuitive and powerful state machine library designed for a great developer experience. Define flat state machines or full statecharts with compound states, parallel regions, and history — all with a clean, pythonic, declarative API that works in both sync and async Python codebases.

Quick start

>>> from statemachine import StateChart, State

>>> class TrafficLightMachine(StateChart):
...     "A traffic light machine"
...     green = State(initial=True)
...     yellow = State()
...     red = State()
...
...     cycle = (
...         green.to(yellow)
...         | yellow.to(red)
...         | red.to(green)
...     )
...
...     def before_cycle(self, event: str, source: State, target: State):
...         return f"Running {event} from {source.id} to {target.id}"
...
...     def on_enter_red(self):
...         print("Don't move.")
...
...     def on_exit_red(self):
...         print("Go ahead!")

Create an instance and send events:

>>> sm = TrafficLightMachine()
>>> sm.send("cycle")
'Running cycle from green to yellow'

>>> sm.send("cycle")
Don't move.
'Running cycle from yellow to red'

>>> sm.send("cycle")
Go ahead!
'Running cycle from red to green'

Check which states are active:

>>> sm.configuration
OrderedSet([State('Green', id='green', value='green', initial=True, final=False, parallel=False)])

>>> sm.green.is_active
True

Generate a diagram or get a text representation with f-strings:

>>> print(f"{sm:md}")
| State  | Event | Guard | Target |
| ------ | ----- | ----- | ------ |
| Green  | Cycle |       | Yellow |
| Yellow | Cycle |       | Red    |
| Red    | Cycle |       | Green  |
<BLANKLINE>
sm._graph().write_png("traffic_light.png")

Parameters are injected into callbacks automatically — the library inspects the signature and provides only the arguments each callback needs:

>>> sm.send("cycle")
'Running cycle from green to yellow'

Guards and conditional transitions

Use cond= and unless= to add guards. When multiple transitions share the same event, declaration order determines priority:

>>> from statemachine import StateChart, State

>>> class ApprovalWorkflow(StateChart):
...     pending = State(initial=True)
...     approved = State(final=True)
...     rejected = State(final=True)
...
...     review = (
...         pending.to(approved, cond="is_valid")
...         | pending.to(rejected)
...     )
...
...     def is_valid(self, score: int = 0):
...         return score >= 70

>>> sm = ApprovalWorkflow()
>>> sm.send("review", score=50)
>>> sm.rejected.is_active
True

>>> sm = ApprovalWorkflow()
>>> sm.send("review", score=85)
>>> sm.approved.is_active
True

The first transition whose guard passes wins. When score < 70, is_valid returns False so the second transition (no guard — always matches) fires instead.

Compound states — hierarchy

Break complex behavior into hierarchical levels with State.Compound. Entering a compound activates both the parent and its initial child. Exiting removes the parent and all descendants:

>>> from statemachine import StateChart, State

>>> class DocumentWorkflow(StateChart):
...     class editing(State.Compound):
...         draft = State(initial=True)
...         review = State()
...         submit = draft.to(review)
...         revise = review.to(draft)
...
...     published = State(final=True)
...     approve = editing.to(published)

>>> sm = DocumentWorkflow()
>>> set(sm.configuration_values) == {"editing", "draft"}
True

>>> sm.send("submit")
>>> "review" in sm.configuration_values
True

>>> sm.send("approve")
>>> set(sm.configuration_values) == {"published"}
True

Parallel states — concurrency

State.Parallel activates all child regions simultaneously. Events in one region don't affect others. A done.state event fires only when all regions reach a final state:

>>> from statemachine import StateChart, State

>>> class DeployPipeline(StateChart):
...     class deploy(State.Parallel):
...         class build(State.Compound):
...             compiling = State(initial=True)
...             compiled = State(final=True)
...             finish_build = compiling.to(compiled)
...         class tests(State.Compound):
...             running = State(initial=True)
...             passed = State(final=True)
...             finish_tests = running.to(passed)
...     released = State(final=True)
...     done_state_deploy = deploy.to(released)

>>> sm = DeployPipeline()
>>> "compiling" in sm.configuration_values and "running" in sm.configuration_values
True

>>> sm.send("finish_build")
>>> "compiled" in sm.configuration_values and "running" in sm.configuration_values
True

>>> sm.send("finish_tests")
>>> set(sm.configuration_values) == {"released"}
True

History states

HistoryState() records which child was active when a compound is exited. Re-entering via the history pseudo-state restores the previous child instead of starting from the initial one:

>>> from statemachine import HistoryState, StateChart, State

>>> class EditorWithHistory(StateChart):
...     class editor(State.Compound):
...         source = State(initial=True)
...         visual = State()
...         h = HistoryState()
...         toggle = source.to(visual) | visual.to(source)
...     settings = State()
...     open_settings = editor.to(settings)
...     back = settings.to(editor.h)

>>> sm = EditorWithHistory()
>>> sm.send("toggle")
>>> "visual" in sm.configuration_values
True

>>> sm.send("open_settings")
>>> sm.send("back")
>>> "visual" in sm.configuration_values
True

Use HistoryState(type="deep") for deep history that remembers the exact leaf state across nested compounds.

Eventless transitions

Transitions without an event trigger fire automatically. With a guard, they fire after any event processing when the condition is met:

>>> from statemachine import StateChart, State

>>> class AutoCounter(StateChart):
...     counting = State(initial=True)
...     done = State(final=True)
...
...     counting.to(done, cond="limit_reached")
...     increment = counting.to.itself(internal=True, on="do_increment")
...
...     count = 0
...
...     def do_increment(self):
...         self.count += 1
...     def limit_reached(self):
...         return self.count >= 3

>>> sm = AutoCounter()
>>> sm.send("increment")
>>> sm.send("increment")
>>> "counting" in sm.configuration_values
True

>>> sm.send("increment")
>>> "done" in sm.configuration_values
True

Error handling

When using StateChart, runtime exceptions in callbacks are caught and turned into error.execution events. Define a transition for that event to handle errors within the state machine itself:

>>> from statemachine import StateChart, State

>>> class ResilientService(StateChart):
...     running = State(initial=True)
...     failed = State(final=True)
...
...     process = running.to(running, on="do_work")
...     error_execution = running.to(failed)
...
...     def do_work(self):
...         raise RuntimeError("something broke")

>>> sm = ResilientService()
>>> sm.send("process")
>>> sm.failed.is_active
True

Async support

Async callbacks just work — same API, no changes needed. The engine detects async callbacks and switches to the async engine automatically:

>>> import asyncio
>>> from statemachine import StateChart, State

>>> class AsyncWorkflow(StateChart):
...     idle = State(initial=True)
...     done = State(final=True)
...
...     finish = idle.to(done)
...
...     async def on_finish(self):
...         return 42

>>> async def run():
...     sm = AsyncWorkflow()
...     result = await sm.finish()
...     print(f"Result: {result}")
...     print(sm.done.is_active)

>>> asyncio.run(run())
Result: 42
True

More features

There's a lot more to explore:

  • DoneData on final states — pass structured data to done.state handlers
  • Delayed events — schedule events with sm.send("event", delay=500)
  • In(state) conditions — cross-region guards in parallel states
  • prepare_event callback — inject custom data into all callbacks
  • Observer pattern — register external listeners to watch events and state changes
  • Django integration — auto-discover state machines in Django apps with MachineMixin
  • Diagram generation — via f-strings (f"{sm:mermaid}"), CLI, Sphinx directive, or Jupyter
  • Dictionary-based definitions — create state machines from data structures
  • Internationalization — error messages in multiple languages

Full documentation: https://python-statemachine.readthedocs.io

Installing

pip install python-statemachine

To generate diagrams, install with the diagrams extra (requires Graphviz):

pip install python-statemachine[diagrams]

Contributing

  • If you found this project helpful, please consider giving it a star on GitHub.

  • Contribute code: If you would like to contribute code, please submit a pull request. For more information on how to contribute, please see our contributing.md file.

  • Report bugs: If you find any bugs, please report them by opening an issue on our GitHub issue tracker.

  • Suggest features: If you have an idea for a new feature, or feel something is harder than it should be, please let us know by opening an issue on our GitHub issue tracker.

  • Documentation: Help improve documentation by submitting pull requests.

  • Promote the project: Help spread the word by sharing on social media, writing a blog post, or giving a talk about it. Tag me on Twitter @fgmacedo so I can share it too!

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

python_statemachine-3.1.2.tar.gz (710.6 kB view details)

Uploaded Source

Built Distribution

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

python_statemachine-3.1.2-py3-none-any.whl (126.1 kB view details)

Uploaded Python 3

File details

Details for the file python_statemachine-3.1.2.tar.gz.

File metadata

  • Download URL: python_statemachine-3.1.2.tar.gz
  • Upload date:
  • Size: 710.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for python_statemachine-3.1.2.tar.gz
Algorithm Hash digest
SHA256 3a64c4ae91d628eb256580f328c68b16e472f25794d2c478e91f2f1dfb8b0668
MD5 211233916b13c11f55cba0655d7c86b8
BLAKE2b-256 afe2dcc389e9855f65a3e54c859499dd409341d5ba561da74af22ce824bc4964

See more details on using hashes here.

Provenance

The following attestation bundles were made for python_statemachine-3.1.2.tar.gz:

Publisher: release.yml on fgmacedo/python-statemachine

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file python_statemachine-3.1.2-py3-none-any.whl.

File metadata

File hashes

Hashes for python_statemachine-3.1.2-py3-none-any.whl
Algorithm Hash digest
SHA256 89dcdbcbf7b197adade16138c75bee4f32e0c98b7eeb18d30aeb1c830dbbdd62
MD5 9dc170853c21632392378912b1b4796f
BLAKE2b-256 f2f5fe9072bfc9d245138561c09bf05b841bf6d92b4ce93d44dbfa18d96e498e

See more details on using hashes here.

Provenance

The following attestation bundles were made for python_statemachine-3.1.2-py3-none-any.whl:

Publisher: release.yml on fgmacedo/python-statemachine

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

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