Overview¶
docs | |
---|---|
tests | |
package |
Minimal state machine
- Free software: BSD license
Usage¶
import fsm
class MyTasks(fsm.FiniteStateMachineMixin):
"""An example to test the state machine.
Contains transitions to everywhere, nowhere and specific states.
"""
state_machine = {
'created': '__all__',
'pending': ('running',),
'running': ('success', 'failed'),
'success': None,
'failed': ('retry',),
'retry': ('pending', 'retry'),
}
def __init__(self, state):
"""Initialize setting a state."""
self.state = state
def on_before_pending(self):
print("I'm going to a pending state")
In [4]: m = MyTasks(state='created')
In [5]: m.change_state('pending')
I'm going to a pending state
Out[5]: 'pending'
In [6]: m.change_state('failed')
---------------------------------------------------------------------------
InvalidTransition Traceback (most recent call last)
<ipython-input-6-71d2461eee74> in <module>()
----> 1 m.change_state('failed')
~/pyfsm/src/fsm/fsm.py in change_state(self, next_state, **kwargs)
90 msg = "The transition from {0} to {1} is not valid".format(previous_state,
91 next_state)
---> 92 raise InvalidTransition(msg)
93
94 name = 'pre_{0}'.format(next_state)
InvalidTransition: The transition from pending to failed is not valid
There are hooks that can be included before a state transition happens and after.
fsm will look for these functions
pre_<state_name>
post_<state_name>
And will give them any extra argument given to change_state
E.g:
Running m.change_state('pending', name='john')
will trigger pre_pending(name='john')
Installation¶
pip install fsmpy
Django integration¶
import fsm
from django.db import models
class MyModel(models.Model, fsm.FiniteStateMachineMixin):
"""An example to test the state machine.
Contains transitions to everywhere, nowhere and specific states.
"""
CHOICES = (
('created', 'CREATED'),
('pending', 'PENDING'),
('running', 'RUNNING'),
('success', 'SUCCESS'),
('failed', 'FAILED'),
('retry', 'RETRY'),
)
state_machine = {
'created': '__all__',
'pending': ('running',),
'running': ('success', 'failed'),
'success': None,
'failed': ('retry',),
'retry': ('pending', 'retry'),
}
state = models.CharField(max_length=30, choices=CHOICES, default='created')
def on_change_state(self, previous_state, next_state, **kwargs):
self.save()
Development¶
To run the tests run:
tox
Note, to combine the coverage data from all the tox environments run:
Windows | set PYTEST_ADDOPTS=--cov-append
tox
|
---|---|
Other | PYTEST_ADDOPTS=--cov-append tox
|