跳转到主要内容

Python中的声明性、明确性、工具友好的有限状态机

项目描述

friendly_states

Build Status Coverage Status Binder

这是一个Python库,用于以易于阅读的方式编写有限状态机,并通过标准linters和IDEs防止错误。

您可以在binder中尝试此README。

介绍

以下是一个简单的声明状态机的示例

from __future__ import annotations
from friendly_states import AttributeState


class TrafficLightMachine(AttributeState):
    is_machine = True

    class Summary:
        Green: [Yellow]
        Yellow: [Red]
        Red: [Green]


class Green(TrafficLightMachine):
    def slow_down(self) -> [Yellow]:
        print("Slowing down...")


class Yellow(TrafficLightMachine):
    def stop(self) -> [Red]:
        print("Stop.")


class Red(TrafficLightMachine):
    def go(self) -> [Green]:
        print("Go!")


TrafficLightMachine.complete()

这看起来像很多代码,但实际上很简单

  1. TrafficLightMachine 声明状态机的根。
  2. 有三个状态,它们是机器的子类:GreenYellowRed
  3. 每个状态通过带返回注释的函数(->)声明它允许到其他状态的转换。在这种情况下,每个状态都有一个到另一个状态的转换。例如,Green可以通过slow_down方法转换到Yellow
  4. TrafficLightMachine.complete() 使机器准备就绪并检查一切是否正确。特别是,它验证TrafficLightMachine内部的Summary类与声明的实际状态和转换相匹配。摘要的每一行都显示了从每个状态的所有可能的输出状态。

正如我们稍后将要看到的,许多样板代码可以为您自动生成,因此不必担心编写所有这些类的努力。

要使用此机器,首先我们需要一个可以存储其自身状态的对象。在使用此库时,您可以根据需要存储状态,但我们的机器声明方式意味着它期望有一个名为 state 的属性。

class TrafficLight:
    def __init__(self, state):
        self.state = state

light = TrafficLight(Green)
assert light.state is Green

当进行状态转换时,就会发生魔法般的变化。

Green(light).slow_down()
Yellow(light).stop()
assert light.state is Red
Slowing down...
Stop.

创建一个如 Green(light) 这样的状态类实例会自动检查 light 是否实际上处于 Green 状态,如果不处于该状态则会引发异常。然后您可以像调用其他类的任何方法一样调用方法。状态将根据方法的返回注解自动更改。

这为什么令人兴奋?因为工具可以清楚地知道每个状态可用的转换。您的IDE可以为您自动补全 Green(light).slo,并且 Green(light).stop() 在您运行代码之前就会被突出显示为错误。

与流行的库 transitions 相比,状态、转换和回调都是使用字符串声明的,因此很容易出错,工具也无法提供帮助。实际上,您必须停止工具警告您关于所有它魔术般生成的属性,您必须使用它们。没有简单的方法可以看到所有转换或从特定状态输出状态。回调可以在转换很远的地方声明。

相比之下,当使用 friendly_states 时,任何地方都没有字符串或魔法属性。代码总是自然地分组在一起:特定于某个状态的所有转换都出现在该类中,与转换相关的逻辑在您期望的地方的函数中。

这里 是使用 friendly_states 实现的 transitions 文档中主示例的等效代码。

基本使用步骤

  1. 请确保您正在使用 Python 3.7+。

  2. 使用 pip install friendly_states 命令安装此库。

  3. 在定义状态机的文件顶部添加一行 from __future__ import annotations

  4. 通过从适当的类(通常是 AttributeState,见 BaseState 部分)继承并在主体中设置 is_machine = True 来声明状态机的根,如下所示

from __future__ import annotations
from friendly_states import AttributeState
    
class MyMachine(AttributeState):
    is_machine = True
    
    class Summary:
        Waiting: [Doing, Finished]
        Doing: [Checking, Finished]
        Checking: [Finished]
        Finished: []

声明摘要不是必须的,但强烈推荐。该类必须命名为 Summary。它应该声明机器将拥有的每个状态,每个状态都注解了可以通过任何转换直接从该状态到达的所有状态列表。

当您调用 MyMachine.complete() 时,将检查此摘要,如果不匹配您的状态类,将引发异常并解释问题。特别是如果某个类缺失,则异常将包含该类的生成源代码,这样您可以花费更少的时间编写样板代码。让我们现在试试

try:
    MyMachine.complete()
except Exception as e:
    print(e)
Missing states:

class Waiting(MyMachine):
    def to_doing(self) -> [Doing]:
        pass

    def to_finished(self) -> [Finished]:
        pass


class Doing(MyMachine):
    def to_checking(self) -> [Checking]:
        pass

    def to_finished(self) -> [Finished]:
        pass


class Checking(MyMachine):
    def to_finished(self) -> [Finished]:
        pass


class Finished(MyMachine):
    pass

您可以复制上面的代码并拥有一个与摘要匹配的工作状态机。这通常不是您需要的,但它应该会为您节省大量时间进行下一步操作。

  1. 为每个状态编写一个类。确保在最后调用 .complete()
class Waiting(MyMachine):
    def start_doing(self) -> [Doing]:
        print('Starting now!')   

    def skip(self) -> [Finished]:
        pass   


class Doing(MyMachine):
    def done(self, result) -> [Checking, Finished]:
        self.obj.result = result
        if self.obj.needs_checking():
            return Checking
        else:
            return Finished


class Checking(MyMachine):
    def check(self) -> [Finished]:
        print('Looks good!')


class Finished(MyMachine):
    pass


MyMachine.complete()

状态类必须(直接或间接)继承自机器类,例如 class Waiting(MyMachine):。一个类可以有任意数量的转换。Waiting 有两个单独的转换,而 Finished 没有转换,这意味着您不能离开该状态。它还可以有任何其他方法或属性,就像普通类一样,不是转换。

转换是指任何具有返回注解(函数定义中->之后的位)的方法,该注解是一个状态列表,这些状态将是此转换的结果。例如,以下代码

class Waiting(MyMachine):
    def start_doing(self) -> [Doing]:

表示start_doing是从状态Waiting转换到状态Doing的转换,调用该方法将改变状态。

转换Doing.done展示了几个有趣的事情

  • 转换可以具有多个可能的输出状态。在这种情况下,该方法必须返回这些状态之一以指示切换到哪个状态。
  • 转换就像普通方法一样,可以接受任何你想要的参数。
  • 状态有一个名为obj的属性,这是在构造类时传递给对象的那个对象。这让你可以与状态正在改变的对象交互。
  1. 机器本身不存储状态,为此需要创建一个不同的类。
class Task:
    def __init__(self):
        self.state = Waiting
        self.result = None
    
    def needs_checking(self):
        return self.result < 5

    
task = Task()
assert task.result is None
assert task.state is Waiting

我们的示例机器期望在对象上找到一个名为state的属性,如下所示。如果你有不同的需求,请参阅BaseState部分。

  1. 要改变对象的状态,你首先需要知道它当前处于什么状态。有时你需要检查,但通常在应用程序的上下文中很明显。例如,如果我们有一个新鲜任务的队列,从这个队列中弹出的任何任务都将处于Waiting状态。

构建正确状态类的实例,并传递你的对象

Waiting(task)
Waiting(obj=<__main__.Task object at 0x1173677f0>)

如果你对任务当前状态的获取有误,这意味着你的代码中存在一个错误!在你可以调用任何转换之前,它将抛出异常。

try:
    Doing(task)
except Exception as e:
    print(e)
<__main__.Task object at 0x1173677f0> should be in state Doing but is actually in state Waiting
  1. 一旦你有正确状态的实例,就像平常一样调用它上面的任何方法。如果方法是转换,状态将自动在之后改变。
Waiting(task).skip()
assert task.state is Finished

如果你尝试调用一个状态不存在的转换,库甚至不需要做任何事情。你将得到你通常在调用不存在的方法时得到的普通Python错误,并且你的IDE/linter会提前警告你。

task = Task()
try:
    Waiting(task).done(3)
except AttributeError as e:
    print(e)
'Waiting' object has no attribute 'done'

以下是任务从等待到完成的另外两个可能的路径

task = Task()
Waiting(task).start_doing()
Doing(task).done(3)
assert task.result == 3
Checking(task).check()
assert task.state is Finished
Starting now!
Looks good!
task = Task()
Waiting(task).start_doing()
Doing(task).done(7)
# The result '7' doesn't need checking
assert task.state is Finished
Starting now!

抽象状态类

有时你将在状态类之间共享常见行为。在这种情况下,你可以使用类继承。以下是你需要注意的事项

  1. 你不能从实际的状态类继承。
  2. 转换必须存在于从机器类(直接或间接)继承的类中。
  3. 继承自机器(因此可以具有转换)但不代表实际状态(因此可以被继承)的类应在其体中具有is_abstract = True

以下是一个例子

class TaskMachine2(AttributeState):
    is_machine = True
    
    class Summary:
        Waiting: [Doing, Finished]
        Doing: [Finished]
        Finished: []

            
class Unfinished(TaskMachine2):
    is_abstract = True
    
    def finish(self) -> [Finished]:
        pass

    
class Waiting(Unfinished):
    def start_doing(self) -> [Doing]:
        pass


class Doing(Unfinished):
    pass


class Finished(TaskMachine2):
    pass


TaskMachine2.complete()    

在这里,WaitingDoing状态都是Unfinished的子类,因此它们可以免费获得finish转换,你可以在总结中看到这个结果。

如果你有一个处于这些状态之一但不确定具体是哪个的对象,并且你想调用finish转换,只需使用Unfinished抽象类即可

import random

for i in range(100):
    task = Task()
    
    # Randomly start doing about half the tasks
    if random.random() < 0.5:
        Waiting(task).start_doing()
    
    # Now the task might be either Waiting or Doing
    Unfinished(task).finish()

但是,如果你尝试在一个已完成的任务上这样做,它将失败

try:
    Unfinished(task).finish()
except Exception as e:
    print(e)
<__main__.Task object at 0x11401cb00> should be in state Unfinished but is actually in state Finished

然而,你可能想使用表示实际当前状态的类中的方法和属性。特别是你可能在具体状态类中覆盖了方法,并希望运行正确的实现。为了允许这样做,状态的实例会自动将类更改为对象的实际状态

task = Task()
assert type(Unfinished(task)) is Waiting

BaseState - 配置状态存储和更改

库中所有类的根是BaseState,它有两个抽象方法get_stateset_state。子类确定对象如何存储状态以及状态改变时应发生什么。

例如,以下是AttributeState的起点,我们迄今为止一直将其用作机器的基础

class AttributeState(BaseState):
    attr_name = "state"

    def get_state(self):
        return getattr(self.obj, self.attr_name)

你可以在机器类中声明不同的attr_name来在对象的该属性中存储状态。

如果您将状态存储在字典或类似的对象中,则可以使用MappingKeyState,其用法如下

class MappingKeyState(BaseState):
    key_name = "state"

    def get_state(self):
        return self.obj[self.key_name]

通常情况下,覆盖set_state以在状态改变时添加额外的公共行为非常有用,例如

class PrintStateChange(AttributeState):
    def set_state(self, previous_state, new_state):
        print(f"Changing {self.obj} from {previous_state} to {new_state}")
        super().set_state(previous_state, new_state)

set_state在转换方法完成后被调用。

因此,您的类层次结构通常如下所示

BaseState <- AttributeState <- Machine <- 抽象状态 <- 实际状态

状态机元数据

机器、状态和转换具有许多您可以检查的属性

# All concrete (not abstract) states in the machine
assert TaskMachine2.states == {Doing, Finished, Waiting}

# All the transition functions available for this state
assert Waiting.transitions == {Waiting.finish, Waiting.start_doing}

# The transition functions defined directly on this class, i.e. not inherited
assert Waiting.direct_transitions == {Waiting.start_doing}

# Possible output states from this transition
assert Waiting.start_doing.output_states == {Doing}

# All possible output states from this state via any transition
assert Waiting.output_states == {Doing, Finished}

# Root of the state machine
assert Waiting.machine is TaskMachine2

# Booleans about the type of class
assert TaskMachine2.is_machine and not Waiting.is_machine
assert Waiting.is_state and not Unfinished.is_state
assert Unfinished.is_abstract and not Waiting.is_abstract

别名和标签

类还具有sluglabel属性,这些属性主要用于Django,但可能在其他地方也有用。slug用于数据存储,而label用于人类显示。

默认情况下,slug只是类名,而label是带有空格的类名。两者都可以通过在类中声明它们来覆盖。

assert Waiting.slug == "Waiting"
assert TaskMachine2.label == "Task Machine 2"

故障排除

如果事情没有按预期工作,以下是一些要检查的事情

  • 检查属性machine.statesstate.transitionstransition.output_states,看看它们是否如预期的那样。
  • 如果您覆盖了set_state,请记得调用super().set_state(...),除非您想阻止状态改变或您直接从BaseState派生。
  • 确保您的转换上的注释是一个列表,即它以[]开始和结束。例如,以下内容将不会被识别为转换
    def start_doing(self) -> Doing:
  • 如果您的转换有任何装饰器,请确保装饰器中仍有原始的__annotations__属性。这通常通过在实现装饰器时使用functools.wraps来完成。
  • 请确保对象以机器期望的方式存储状态。通常您会使用AttributeState,并且您应该确保attr_name(默认为“state”)是正确的。请注意,典型的机器期望对象只以一种方式存储状态 - 您不能使用相同的机器来更改存储在不同属性中的状态。要克服这一点,请参阅“动态更改属性名”配方
  • 检查您是否正确地继承了类。所有状态都需要从机器继承。

食谱

friendly_states的API故意很简单。以下是您可以进行一些更复杂操作的方法。

构建和绘制图形

以下是如何使用流行的库networkx创建图表

import networkx as nx
machine = MyMachine

G = nx.DiGraph()
for state in machine.states:
    for output_state in state.output_states:
        G.add_edge(state, output_state)

print(G.nodes)
print(G.edges)
[Waiting, Finished, Doing, Checking]
[(Waiting, Finished), (Waiting, Doing), (Doing, Finished), (Doing, Checking), (Checking, Finished)]

以下是如何使用matplotlib绘制图表

%matplotlib inline
nx.draw(G, with_labels=True, node_color='pink')   

png

要为每条边添加标签需要更多的工作

edge_labels = {}
G = nx.DiGraph()
for state in machine.states:
    for transition in state.transitions:
        for output_state in transition.output_states:
            edge = (state, output_state)
            G.add_edge(*edge)
            edge_labels[edge] = transition.__name__

pos = nx.spring_layout(G)
nx.draw(G, pos, with_labels=True, node_color='pink')
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels);

png

创建多个类似机器

假设您想创建具有类似状态和转换的多个机器,而无需重复代码。您可能会认为您可以用某种方式使用继承,但这不会起作用,实际上机器不能从另一个机器派生。相反,您必须创建一个本地创建类的函数。这取决于您的需求有多种方法。以下是一个示例,其中机器完全相同,只有一个额外的状态

from types import SimpleNamespace

def machine_factory():
    class Machine(AttributeState):
        is_machine = True

    class CommonState1(Machine):
        def to_common_state_2(self) -> [CommonState2]:
            pass

    class CommonState2(Machine):
        pass

    return SimpleNamespace(
        Machine=Machine,
        CommonState1=CommonState1,
        CommonState2=CommonState2,
    )


machine1 = machine_factory()

class DifferentState(machine1.Machine):
    def to_common_state_2(self) -> [machine1.CommonState2]:
        pass

machine1.Machine.complete()

@machine1.Machine.check_summary
class Summary:
    CommonState1: [CommonState2]
    CommonState2: []
    DifferentState: [CommonState2]


machine2 = machine_factory()
machine2.Machine.complete()

@machine2.Machine.check_summary
class Summary:
    CommonState1: [CommonState2]
    CommonState2: []

动态更改属性名称

典型的AttributeState机器只能使用一个属性名。您可能需要使用相同的机器与使用不同属性的不同的对象类。一种方法是使用与上面类似的可配置属性名的模式

def machine_factory(name):
    class Machine(AttributeState):
        is_machine = True
        attr_name = name

    ...

另一种可能适用于更复杂情况的选项是,通过在构建时接受属性名来派生AttributeState

class DynamicAttributeState(AttributeState):
    def __init__(self, obj, attr_name):
        # override the class attribute
        self.attr_name = attr_name
        
        # must call super *after* because it checks the state
        # in the attribute with the given name
        super().__init__(obj)

class Machine(DynamicAttributeState):
    is_machine = True

class Start(Machine):
    def to_end(self) -> [End]:
        pass

class End(Machine):
    pass

Machine.complete()

thing = SimpleNamespace(state=Start, other_state=Start)

assert thing.state is Start
assert thing.other_state is Start

Start(thing, "state").to_end()
assert thing.state is End
assert thing.other_state is Start

Start(thing, "other_state").to_end()
assert thing.state is End
assert thing.other_state is End

进入/退出状态回调

每次您想在每次状态转换上执行一些通用逻辑时,都应该覆盖set_state。但如果您将所有代码都放在那里,它可能会变得很长且难以理解。如果您想将此逻辑分组到您的状态类中,以便在转换进入或退出该状态时执行,这里有一个您可以应用于任何机器的mixin

class OnEnterExitMixin:
    def set_state(self, previous_state, new_state):
        previous_state(self.obj).on_exit(new_state)
        super().set_state(previous_state, new_state)
        new_state(self.obj).on_enter(previous_state)

    def on_exit(self, new_state):
        pass

    def on_enter(self, previous_state):
        pass

然后按照以下方式使用它

class Machine(OnEnterExitMixin, AttributeState):
    is_machine = True

class Start(Machine):
    def end(self) -> [End]:
        pass

class End(Machine):
    def on_enter(self, previous_state):
        print(f"Ending from {previous_state}")

Machine.complete()

thing = SimpleNamespace(state=Start)
Start(thing).end()
Ending from Start

Django集成

friendly_states可以轻松地与Django一起使用。基本用法如下

from django.db import models
from friendly_states.django import StateField, DjangoState

class MyMachine(DjangoState):
    is_machine = True
    
# ...

class MyModel(models.Model):
    state = StateField(MyMachine)

StateField 是一个 CharField,它在数据库中存储当前状态的 slug,同时允许你在所有代码中使用实际的状态类对象,例如:

obj = MyModel.objects.create(state=MyState)
assert obj.state is MyState
objects = MyModel.objects.filter(state=MyState)

所有关键字参数都会直接传递给 CharField,除了 max_lengthchoices,它们将被忽略,见下文。

DjangoState 将在状态转换后自动保存您的模型。要禁用此功能,请在您的机器或状态类上设置 auto_save = False

StateField 将自动在模型中找到其名称,并在机器上设置相应的 attr_name,因此您不需要设置它。但请注意,您不能为同一机器使用不同的属性名称。

因为数据库存储的是 slugs,而默认情况下 slug 是类名,如果您在代码中重命名了类,并且希望现有数据保持有效,则应将 slug 设置为旧类名

class MyRenamedState(MyMachine):
    slug = "MyState"
    ...

同样,只要数据库中包含该状态的对象,就不能删除状态类,否则在尝试使用此类对象时,您的代码将失败。

max_length 将自动设置为机器中所有 slugs 的最大长度。如果您想节省数据库空间,可以覆盖 slugs 以缩短它们。

choices 由每个状态的字段 sluglabel 构成。要自定义状态在表单等中的显示方式,请覆盖类上的 label 属性。

项目详情


下载文件

下载适用于您平台文件。如果您不确定选择哪个,请了解有关 安装包 的更多信息。

源分布

friendly_states-0.2.0.tar.gz (24.0 kB 查看散列值)

上传时间

由以下机构支持: