最小状态机
项目描述
docs |
|
---|---|
tests |
|
package |
最小状态机
自由软件:BSD许可
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
安装
pip install fsmpy
使用
在类中定义
state_machine
初始化
state
,可以通过值、使用__init__
或作为django字段添加钩子
方法 |
描述 |
on_before_change_state |
在转换到状态之前 |
on_change_state |
转换到状态后(如果没有失败),对每个状态都会运行 |
pre_<state_name> |
在特定状态之前运行,其中 |
post_<state_name> |
在特定状态之后运行,其中 |
这些钩子将接收传递给change_state
的任何额外参数
例如
运行m.change_state('pending', name='john')
将触发pre_pending(name='john')
Django集成
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
如果您使用serializers
,它们通常执行save
操作,因此在on_change_state
中保存是不必要的。
一个简单的解决方案是这样做
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
文档
开发
要运行测试,请执行
tox
注意,要合并来自所有tox环境的覆盖率数据,请执行
Windows |
set PYTEST_ADDOPTS=--cov-append tox |
---|---|
其他 |
PYTEST_ADDOPTS=--cov-append tox |
项目详情
下载文件
下载适合您平台的文件。如果您不确定选择哪个,请了解有关安装包的更多信息。
源分布
fsmpy-2.1.0.tar.gz (6.0 kB 查看哈希值)
构建分布
fsmpy-2.1.0-py2.py3-none-any.whl (5.8 kB 查看哈希值)
关闭
fsmpy-2.1.0.tar.gz 的哈希值
算法 | 哈希摘要 | |
---|---|---|
SHA256 | be81b809d1ce4cf7647f4293897e9656e2e8b12f7f303f5b678ac921c3bbfb4d |
|
MD5 | 9cdfcc14e460513b19cd52e44c68847f |
|
BLAKE2b-256 | 25b03f7229c149d2fcf6b93fb42d67e2e22ba023b1f4859b837e2b1d569117de |
关闭
fsmpy-2.1.0-py2.py3-none-any.whl 的哈希值
算法 | 哈希摘要 | |
---|---|---|
SHA256 | fb16c0f5376afa868b1946c90205a2ff8fd877a7d8636e2574b5b6b6823fdd99 |
|
MD5 | e2dfcc9d794f24453d9196d682f7a002 |
|
BLAKE2b-256 | 51e0f579621301175dfe016dcb7678e83a09ca58e8bb96882d05094d37d78454 |