Home

Awesome

Python StateMachine

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

Python finite-state machines made easy.

<div align="center">

</div>

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

Features

Installing

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

pip install python-statemachine

To generate diagrams from your machines, you'll also need pydot and Graphviz. You can install this library already with pydot dependency using the extras install option. See our docs for more details.

pip install python-statemachine[diagrams]

First example

Define your state machine:

>>> from statemachine import StateMachine, State

>>> class TrafficLightMachine(StateMachine):
...     "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, message: str = ""):
...         message = ". " + message if message else ""
...         return f"Running {event} from {source.id} to {target.id}{message}"
...
...     def on_enter_red(self):
...         print("Don't move.")
...
...     def on_exit_red(self):
...         print("Go ahead!")

You can now create an instance:

>>> sm = TrafficLightMachine()

This state machine can be represented graphically as follows:

>>> img_path = "docs/images/readme_trafficlightmachine.png"
>>> sm._graph().write_png(img_path)

Where on the TrafficLightMachine, we've defined green, yellow, and red as states, and one event called cycle, which is bound to the transitions from green to yellow, yellow to red, and red to green. We also have defined three callbacks by name convention, before_cycle, on_enter_red, and on_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_state
State('Yellow', id='yellow', value='yellow', initial=False, final=False)

The State instance can also be checked by equality:

>>> sm.current_state == TrafficLightMachine.yellow
True

>>> sm.current_state == sm.yellow
True

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

>>> sm.green.is_active
False

>>> sm.yellow.is_active
True

>>> sm.red.is_active
False

Easily iterate over all states:

>>> [s.id for s in sm.states]
['green', 'red', 'yellow']

Or over events:

>>> [t.name for t in sm.events]
['cycle']

Call an event by its name:

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

Or send an event with the event name:

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

>>> sm.green.is_active
True

You can pass arbitrary positional or keyword arguments to the event, and they 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 the callback method.

Note how before_cycle was declared:

def before_cycle(self, event: str, source: State, target: State, message: str = ""):
    message = ". " + message if message else ""
    return f"Running {event} from {source.id} to {target.id}{message}"

The params event, source, target (and others) are available built-in to be used on any action. The param message is user-defined, in our example we made it default empty so we can call cycle with or without a message parameter.

If we pass a message parameter, it will be used on the before_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 events raise a TransitionNotAllowed exception:

>>> sm.send("go")
Traceback (most recent call last):
statemachine.exceptions.TransitionNotAllowed: Can't go when in Yellow.

Keeping the same state as expected:

>>> sm.yellow.is_active
True

A human-readable name is automatically derived from the State.id, which is used on the messages and in diagrams:

>>> sm.current_state.name
'Yellow'

Async support

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

>>> class AsyncStateMachine(StateMachine):
...     initial = State('Initial', initial=True)
...     final = State('Final', final=True)
...
...     advance = initial.to(final)
...
...     async def on_advance(self):
...         return 42

>>> async def run_sm():
...     sm = AsyncStateMachine()
...     result = await sm.advance()
...     print(f"Result is {result}")
...     print(sm.current_state)

>>> asyncio.run(run_sm())
Result is 42
Final

A more useful example

A simple didactic state machine for controlling an Order:

>>> class OrderControl(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__()
...
...     def payments_enough(self, amount):
...         return sum(self.payments) + amount >= self.order_total
...
...     def before_add_to_order(self, amount):
...         self.order_total += amount
...         return self.order_total
...
...     def before_receive_payment(self, amount):
...         self.payments.append(amount)
...         return self.payments
...
...     def after_receive_payment(self):
...         self.payment_received = True
...
...     def on_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 (most recent call last):
...
statemachine.exceptions.TransitionNotAllowed: Can't process_order when in Waiting for payment.

>>> control.receive_payment(6)
[4, 6]

>>> control.current_state.id
'processing'

>>> control.process_order()

>>> control.ship_order()

>>> control.payment_received
True

>>> control.order_total
10

>>> control.payments
[4, 6]

>>> control.completed.is_active
True

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

Contributing