跳转到主要内容

最小状态机

项目描述

docs

Read the Docs

tests

Travis-CI Build Status
Coverage Status

package

PyPI Package latest release PyPI Wheel Supported versions Supported implementations

最小状态机

  • 自由软件: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

使用

  1. 在类中定义state_machine

  2. 初始化state,可以通过值、使用__init__或作为django字段

  3. 添加钩子

方法

描述

on_before_change_state

在转换到状态之前

on_change_state

转换到状态后(如果没有失败),对每个状态都会运行

pre_<state_name>

在特定状态之前运行,其中state_namestate_machine中指定的名称

post_<state_name>

在特定状态之后运行,其中state_namestate_machine中指定的名称

这些钩子将接收传递给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

文档

https://pyfsm.readthedocs.org/

开发

要运行测试,请执行

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 查看哈希值)

上传时间 Python 2 Python 3

由以下支持

AWSAWS 云计算和安全赞助商 DatadogDatadog 监控 FastlyFastly CDN GoogleGoogle 下载分析 MicrosoftMicrosoft PSF赞助商 PingdomPingdom 监控 SentrySentry 错误日志 StatusPageStatusPage 状态页面