Python StateMachine

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

>>>fromstatemachineimportStateChart, State>>>classTrafficLightMachine(StateChart):
... "A traffic light machine"
... green=State(initial=True)
... yellow=State()
... red=State()
...
... cycle= (
... green.to(yellow)
... |yellow.to(red)
... |red.to(green)
... )
...
... defbefore_cycle(self, event: str, source: State, target: State):
... returnf"Running {event} from {source.id} to {target.id}"
...
... defon_enter_red(self):
... print("Don't move.")
...
... defon_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'tmove.
'Running cycle from yellow to red'>>>sm.send("cycle")
Goahead!
'Running cycle from red to green'

Check which states are active:

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

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:

>>>fromstatemachineimportStateChart, State>>>classApprovalWorkflow(StateChart):
... pending=State(initial=True)
... approved=State(final=True)
... rejected=State(final=True)
...
... review= (
... pending.to(approved, cond="is_valid")
... |pending.to(rejected)
... )
...
... defis_valid(self, score: int=0):
... returnscore>=70>>>sm=ApprovalWorkflow()
>>>sm.send("review", score=50)
>>>sm.rejected.is_activeTrue>>>sm=ApprovalWorkflow()
>>>sm.send("review", score=85)
>>>sm.approved.is_activeTrue

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:

>>>fromstatemachineimportStateChart, State>>>classDocumentWorkflow(StateChart):
... classediting(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"insm.configuration_valuesTrue>>>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:

>>>fromstatemachineimportStateChart, State>>>classDeployPipeline(StateChart):
... classdeploy(State.Parallel):
... classbuild(State.Compound):
... compiling=State(initial=True)
... compiled=State(final=True)
... finish_build=compiling.to(compiled)
... classtests(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"insm.configuration_valuesand"running"insm.configuration_valuesTrue>>>sm.send("finish_build")
>>>"compiled"insm.configuration_valuesand"running"insm.configuration_valuesTrue>>>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:

>>>fromstatemachineimportHistoryState, StateChart, State>>>classEditorWithHistory(StateChart):
... classeditor(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"insm.configuration_valuesTrue>>>sm.send("open_settings")
>>>sm.send("back")
>>>"visual"insm.configuration_valuesTrue

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:

>>>fromstatemachineimportStateChart, State>>>classAutoCounter(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
...
... defdo_increment(self):
... self.count+=1
... deflimit_reached(self):
... returnself.count>=3>>>sm=AutoCounter()
>>>sm.send("increment")
>>>sm.send("increment")
>>>"counting"insm.configuration_valuesTrue>>>sm.send("increment")
>>>"done"insm.configuration_valuesTrue

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:

>>>fromstatemachineimportStateChart, State>>>classResilientService(StateChart):
... running=State(initial=True)
... failed=State(final=True)
...
... process=running.to(running, on="do_work")
... error_execution=running.to(failed)
...
... defdo_work(self):
... raiseRuntimeError("something broke")
>>>sm=ResilientService()
>>>sm.send("process")
>>>sm.failed.is_activeTrue

Async support

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

>>>importasyncio>>>fromstatemachineimportStateChart, State>>>classAsyncWorkflow(StateChart):
... idle=State(initial=True)
... done=State(final=True)
...
... finish=idle.to(done)
...
... asyncdefon_finish(self):
... return42>>>asyncdefrun():
... sm=AsyncWorkflow()
... result=awaitsm.finish()
... print(f"Result: {result}")
... print(sm.done.is_active)
>>>asyncio.run(run())
Result: 42True

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]

To load statecharts from declarative documents, install the IO extras (yaml for YAML, validation for validate=True, or io for both):

pip install python-statemachine[io]

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!