Minimal state machine
Project description
docs | |
---|---|
tests | |
package |
Minimal state machine
- Free software: BSD license
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') # Let's try to transition to an invalid state --------------------------------------------------------------------------- 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
Installation
pip install fsmpy
Usage
- Define in a class the
state_machine
- Initialize
state
, either with a value, using__init__
or as a django field - Add hooks:
Method | Description |
on_before_change_state | Before transitioning to the state |
on_change_state | After transitioning to the state, if no failure, runs for every state |
pre_<state_name> | Runs before a particular state, where state_name is the specified name in the state_machine |
post_<state_name> | Runs after a particular state, where state_name is the specified name in the state_machine |
This hooks will receive any extra argument given to change_state
E.g:
Running m.change_state('pending', name='john')
will trigger pre_pending(name='john')
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()
Django Rest Framework
If you are using serializers
, they usually perform the save
, so saving inside on_change_state
is not necessary.
One simple solution is to do this:
class MySerializer(serializers.ModelSerializer): def update(self, instance, validated_data): new_state = validated_data.get('state', instance.state) try: instance.change_state(new_state) except fsm.InvalidTransition: raise serializers.ValidationError("Invalid transition") instance = super().update(instance, validated_data) return instance
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 |
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
fsmpy-2.1.0.tar.gz
(6.0 kB
view hashes)
Built Distribution
Close
Hashes for fsmpy-2.1.0-py2.py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | fb16c0f5376afa868b1946c90205a2ff8fd877a7d8636e2574b5b6b6823fdd99 |
|
MD5 | e2dfcc9d794f24453d9196d682f7a002 |
|
BLAKE2-256 | 51e0f579621301175dfe016dcb7678e83a09ca58e8bb96882d05094d37d78454 |