Python中的声明性、明确性、工具友好的有限状态机
项目描述
friendly_states
这是一个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()
这看起来像很多代码,但实际上很简单
TrafficLightMachine
声明状态机的根。- 有三个状态,它们是机器的子类:
Green
、Yellow
和Red
。 - 每个状态通过带返回注释的函数(
->
)声明它允许到其他状态的转换。在这种情况下,每个状态都有一个到另一个状态的转换。例如,Green
可以通过slow_down
方法转换到Yellow
。 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
文档中主示例的等效代码。
基本使用步骤
-
请确保您正在使用 Python 3.7+。
-
使用
pip install friendly_states
命令安装此库。 -
在定义状态机的文件顶部添加一行
from __future__ import annotations
。 -
通过从适当的类(通常是
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
您可以复制上面的代码并拥有一个与摘要匹配的工作状态机。这通常不是您需要的,但它应该会为您节省大量时间进行下一步操作。
- 为每个状态编写一个类。确保在最后调用
.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
的属性,这是在构造类时传递给对象的那个对象。这让你可以与状态正在改变的对象交互。
- 机器本身不存储状态,为此需要创建一个不同的类。
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
部分。
- 要改变对象的状态,你首先需要知道它当前处于什么状态。有时你需要检查,但通常在应用程序的上下文中很明显。例如,如果我们有一个新鲜任务的队列,从这个队列中弹出的任何任务都将处于
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
- 一旦你有正确状态的实例,就像平常一样调用它上面的任何方法。如果方法是转换,状态将自动在之后改变。
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!
抽象状态类
有时你将在状态类之间共享常见行为。在这种情况下,你可以使用类继承。以下是你需要注意的事项
- 你不能从实际的状态类继承。
- 转换必须存在于从机器类(直接或间接)继承的类中。
- 继承自机器(因此可以具有转换)但不代表实际状态(因此可以被继承)的类应在其体中具有
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()
在这里,Waiting
和Doing
状态都是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_state
和set_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
别名和标签
类还具有slug
和label
属性,这些属性主要用于Django,但可能在其他地方也有用。slug
用于数据存储,而label
用于人类显示。
默认情况下,slug
只是类名,而label
是带有空格的类名。两者都可以通过在类中声明它们来覆盖。
assert Waiting.slug == "Waiting"
assert TaskMachine2.label == "Task Machine 2"
故障排除
如果事情没有按预期工作,以下是一些要检查的事情
- 检查属性
machine.states
、state.transitions
和transition.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')
要为每条边添加标签需要更多的工作
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);
创建多个类似机器
假设您想创建具有类似状态和转换的多个机器,而无需重复代码。您可能会认为您可以用某种方式使用继承,但这不会起作用,实际上机器不能从另一个机器派生。相反,您必须创建一个本地创建类的函数。这取决于您的需求有多种方法。以下是一个示例,其中机器完全相同,只有一个额外的状态
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_length
和 choices
,它们将被忽略,见下文。
DjangoState
将在状态转换后自动保存您的模型。要禁用此功能,请在您的机器或状态类上设置 auto_save = False
。
StateField
将自动在模型中找到其名称,并在机器上设置相应的 attr_name
,因此您不需要设置它。但请注意,您不能为同一机器使用不同的属性名称。
因为数据库存储的是 slugs,而默认情况下 slug 是类名,如果您在代码中重命名了类,并且希望现有数据保持有效,则应将 slug 设置为旧类名
class MyRenamedState(MyMachine):
slug = "MyState"
...
同样,只要数据库中包含该状态的对象,就不能删除状态类,否则在尝试使用此类对象时,您的代码将失败。
max_length
将自动设置为机器中所有 slugs 的最大长度。如果您想节省数据库空间,可以覆盖 slugs 以缩短它们。
choices
由每个状态的字段 slug
和 label
构成。要自定义状态在表单等中的显示方式,请覆盖类上的 label
属性。
项目详情
friendly_states-0.2.0.tar.gz 的散列值
算法 | 散列摘要 | |
---|---|---|
SHA256 | 981be61f82455cbdd8a281f2ea0d068ebceb76002824f984d816369847731eda |
|
MD5 | 4ccdda28dd0b512fd74f35fc5e888edf |
|
BLAKE2b-256 | a9a466accf3e80247c1bbf8e96fcacb159bf07db3d6e28af75ea45085786e81d |