Repository files navigation

Pythonfinite-state machines made easy.

Welcome to python-statemachine, an intuitive and powerful state machine library designed for agreat developer experience. We provide apythonic and expressive API for implementing statemachines in sync or asynchonous Python codebases.

• ✨Basic components: Easily defineStates,Events, andTransitions to model your logic.
• ⚙️Actions and handlers: Attach actions and handlers to states, events, and transitions to control behavior dynamically.
• 🛡️Conditional transitions: ImplementGuards andValidators to conditionally control transitions, ensuring they only occur when specific conditions are met.
• 🚀Full async support: Enjoy full asynchronous support. Await events, and dispatch callbacks asynchronously for seamless integration with async codebases.
• 🔄Full sync support: Use the same state machine from synchronous codebases without any modifications.
• 🎨Declarative and simple API: Utilize a clean, elegant, and readable API to define your state machine, making it easy to maintain and understand.
• 👀Observer pattern support: Register external and generic objects to watch events and register callbacks.
• 🔍Decoupled design: Separate concerns with a decoupled "state machine" and "model" design, promoting cleaner architecture and easier maintenance.
• ✅Correctness guarantees: Ensured correctness with validations at class definition time:
  • Ensures exactly oneinitial state.
  • Disallows transitions fromfinal states.
  • Requires ongoing transitions for all non-final states.
  • Guarantees all non-final states have at least one path to a final state if final states are declared.
  • Validates the state machine graph representation has a single component.
• 📦Flexible event dispatching: Dispatch events with any extra data, making it available to all callbacks, including actions and guards.
• 🔧Dependency injection: Needed parameters are injected into callbacks.
• 📊Graphical representation: Generate and output graphical representations of state machines. Create diagrams from the command line, at runtime, or even in Jupyter notebooks.
• 🌍Internationalization support: Provides error messages in different languages, making the library accessible to a global audience.
• 🛡️Robust testing: Ensured reliability with a codebase that is 100% covered by automated tests, including all docs examples. Releases follow semantic versioning for predictable releases.
• 🏛️Domain model integration: Seamlessly integrate with domain models using Mixins.
• 🔧Django integration: Automatically discover state machines in Django applications.

To install Python State Machine, run this command in your terminal:

pip install python-statemachine

To generate diagrams from your machines, you'll also needpydot andGraphviz. You caninstall this library already withpydot dependency using theextras install option. Seeour docs for more details.

pip install python-statemachine[diagrams]

Define your state machine:

>>>fromstatemachineimportStateMachine,State>>>classTrafficLightMachine(StateMachine):..."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,message:str=""):...message=". "+messageifmessageelse""...returnf"Running{event} from{source.id} to{target.id}{message}"......defon_enter_red(self):...print("Don't move.")......defon_exit_red(self):...print("Go ahead!")

You can now create an instance:

>>>sm=TrafficLightMachine()

This state machine can be represented graphically as follows:

>>># This example will only run on automated tests if dot is present>>>getfixture("requires_dot_installed")>>>img_path="docs/images/readme_trafficlightmachine.png">>>sm._graph().write_png(img_path)

Where on theTrafficLightMachine, we've definedgreen,yellow, andred as states, andone event calledcycle, which is bound to the transitions fromgreen toyellow,yellow tored,andred togreen. We also have defined three callbacks by name convention,before_cycle,on_enter_red, andon_exit_red.

Then start sending events to your new state machine:

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

That's it. This is all an external object needs to know about your state machine: How to send events.Ideally, all states, transitions, and actions should be kept internally and not checked externally to avoid unnecessary coupling.

But if your use case needs, you can inspect state machine properties, like the current state:

>>>sm.current_state.id'yellow'

Or get a complete state representation for debugging purposes:

>>>sm.current_stateState('Yellow',id='yellow',value='yellow',initial=False,final=False)

TheState instance can also be checked by equality:

>>>sm.current_state==TrafficLightMachine.yellowTrue>>>sm.current_state==sm.yellowTrue

Or you can check if a state is active at any time:

>>>sm.green.is_activeFalse>>>sm.yellow.is_activeTrue>>>sm.red.is_activeFalse

Easily iterate over all states:

>>> [s.idforsinsm.states]['green','yellow','red']

Or over events:

>>> [t.idfortinsm.events]['cycle']

Call an event by its id:

>>>sm.cycle()Don'tmove.'Running cycle from yellow to red'

Or send an event with the event id:

>>>sm.send('cycle')Goahead!'Running cycle from red to green'>>>sm.green.is_activeTrue

You can pass arbitrary positional or keyword arguments to the event, andthey will be propagated to all actions and callbacks using something similar to dependency injection. In other words, the library will only inject the parameters declared on thecallback method.

Note howbefore_cycle was declared:

defbefore_cycle(self,event:str,source:State,target:State,message:str=""):message=". "+messageifmessageelse""returnf"Running{event} from{source.id} to{target.id}{message}"

The paramsevent,source,target (and others) are available built-in to be used on any action.The parammessage is user-defined, in our example we made it default empty so we can callcycle withor without amessage parameter.

If we pass amessage parameter, it will be used on thebefore_cycle action:

>>>sm.send("cycle",message="Please, now slowdown.")'Running cycle from green to yellow. Please, now slowdown.'

By default, events with transitions that cannot run from the current state or unknown eventsraise aTransitionNotAllowed exception:

>>>sm.send("go")Traceback (mostrecentcalllast):statemachine.exceptions.TransitionNotAllowed:Can'tgowheninYellow.

Keeping the same state as expected:

>>>sm.yellow.is_activeTrue

A human-readable name is automatically derived from theState.id, which is used on the messagesand in diagrams:

>>>sm.current_state.name'Yellow'

We support native coroutine usingasyncio, enabling seamless integration with asynchronous code.There's no change on the public API of the library to work on async codebases.

>>>classAsyncStateMachine(StateMachine):...initial=State('Initial',initial=True)...final=State('Final',final=True)......advance=initial.to(final)......asyncdefon_advance(self):...return42>>>asyncdefrun_sm():...sm=AsyncStateMachine()...result=awaitsm.advance()...print(f"Result is{result}")...print(sm.current_state)>>>asyncio.run(run_sm())Resultis42Final

A simple didactic state machine for controlling anOrder:

>>>classOrderControl(StateMachine):...waiting_for_payment=State(initial=True)...processing=State()...shipping=State()...completed=State(final=True)......add_to_order=waiting_for_payment.to(waiting_for_payment)...receive_payment= (...waiting_for_payment.to(processing,cond="payments_enough")...|waiting_for_payment.to(waiting_for_payment,unless="payments_enough")...     )...process_order=processing.to(shipping,cond="payment_received")...ship_order=shipping.to(completed)......def__init__(self):...self.order_total=0...self.payments= []...self.payment_received=False...super(OrderControl,self).__init__()......defpayments_enough(self,amount):...returnsum(self.payments)+amount>=self.order_total......defbefore_add_to_order(self,amount):...self.order_total+=amount...returnself.order_total......defbefore_receive_payment(self,amount):...self.payments.append(amount)...returnself.payments......defafter_receive_payment(self):...self.payment_received=True......defon_enter_waiting_for_payment(self):...self.payment_received=False

You can use this machine as follows.

>>>control=OrderControl()>>>control.add_to_order(3)3>>>control.add_to_order(7)10>>>control.receive_payment(4)[4]>>>control.current_state.id'waiting_for_payment'>>>control.current_state.name'Waiting for payment'>>>control.process_order()Traceback (mostrecentcalllast):...statemachine.exceptions.TransitionNotAllowed:Can'tprocess_orderwheninWaitingforpayment.>>>control.receive_payment(6)[4,6]>>>control.current_state.id'processing'>>>control.process_order()>>>control.ship_order()>>>control.payment_receivedTrue>>>control.order_total10>>>control.payments[4,6]>>>control.completed.is_activeTrue

There's a lot more to cover, please take a look at our docs:https://python-statemachine.readthedocs.io.

• Star this project
• Open an Issue
• Fork

• 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 pullrequest. For more information on how to contribute, please see ourcontributing.md file.

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

• Suggest features: If you have an idea for a new feature, of feels something being 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!