政治家

用现代 Python 构建状态机的外交途径。

statesman 是一个库，它提供了一个优雅且富有表现力的 API，用于在异步 Python 3.8+ 中实现状态机。它将代表您与复杂性进行谈判，并就未来如何管理状态达成一个清晰、简洁的协议。

特征

• 有限状态机库（状态、事件、转换、动作）中常见问题的轻量级但功能齐全的实现。
• 使用类型提示、装饰器和枚举的声明性简单 API。
• 为状态和事件提供一组丰富的操作和回调。状态支持进入和退出动作，事件有保护、之前、开始和之后动作。状态机范围的回调是通过可覆盖的方法提供的。
• 设计和构建异步原生。回调是异步分派的，因此很容易与长时间运行的事件驱动流程集成。
• 防护操作可以在应用任何状态更改之前取消转换。
• 数据可以直接在 Pydantic的状态机子类上建模。
• 事件支持使用可用于所有操作的任意关联参数数据。参数与接收可调用的签名匹配，从而实现关注点的划分。
• 可靠的测试覆盖率和文档。

例子

那么……它是什么样子的？很高兴你问。

from typing import Optional, List
import statesman


class ProcessLifecycle(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = "Starting..."
        running = "Running..."
        stopping = "Stopping..."
        stopped = "Terminated."

    # Track state about the process we are running
    command: Optional[str] = None
    pid: Optional[int] = None
    logs: List[str] = []

    # initial state entry point
    @statesman.event(None, States.starting)
    async def start(self, command: str) -> None:
        """"Start a process."""
        self.command = command
        self.pid = 31337
        self.logs.clear()  # Flush logs between runs

    @statesman.event(source=States.starting, target=States.running)
    async def run(self, transition: statesman.Transition) -> None:
        """Mark the process as running."""
        self.logs.append(f"Process pid {self.pid} is now running (command=\"{self.command}\")")

    @statesman.event(source=States.running, target=States.stopping)
    async def stop(self) -> None:
        """Stop a running process."""
        self.logs.append(f"Shutting down pid {self.pid} (command=\"{self.command}\")")

    @statesman.event(source=States.stopping, target=States.stopped)
    async def terminate(self) -> None:
        """Terminate a running process."""
        self.logs.append(f"Terminated pid {self.pid} (\"{self.command}\")")
        self.command = None
        self.pid = None

    @statesman.enter_state(States.stopping)
    async def _print_status(self) -> None:
        print("Entering stopped status!")

    @statesman.after_event("run")
    async def _after_run(self) -> None:
        print("running...")

    async def after_transition(self, transition: statesman.Transition) -> None:
        if transition.event and transition.event.name == "stop":
            await self.terminate()


async def _examples():
    # Let's play.
    state_machine = ProcessLifecycle()
    await state_machine.start("ls -al")
    assert state_machine.command == "ls -al"
    assert state_machine.pid == 31337
    assert state_machine.state == ProcessLifecycle.States.starting

    await state_machine.run()
    assert state_machine.logs == ['Process pid 31337 is now running (command="ls -al")']

    await state_machine.stop()
    assert state_machine.logs == [
        'Process pid 31337 is now running (command="ls -al")',
        'Shutting down pid 31337 (command="ls -al")',
        'Terminated pid 31337 ("ls -al")',
    ]

    # Or start in a specific state
    state_machine = ProcessLifecycle(state=ProcessLifecycle.States.running)

    # Transition to a specific state
    await state_machine.enter_state(ProcessLifecycle.States.stopping)

    # Trigger an event
    await state_machine.trigger_event("stop", key="value")

状态被定义为 Python 枚举类。枚举项的名称定义状态的符号名称，值提供人类可读的描述。一个名为States嵌入在状态机子类中的类会自动绑定到状态机。

使用事件装饰器声明事件并定义转换的源和目标状态。源状态None定义了初始状态转换。

一旦方法被装饰为事件操作，原始方法主体将作为 on 事件操作附加到新事件，并且该方法被替换为触发新创建事件的实现。

动作可以在声明时或稍后通过 guard_event、before_event、on_event和after_event装饰器附加到事件。动作同样可以通过enter_state和exit_state 装饰器附加到状态。

有一个广泛的 API 可用于以编程方式处理状态机及其组件。

为什么是政治家？

政治家之所以被开发，是因为我们找不到类似的东西。虽然 Python 社区在 FSM 库方面存在着令人尴尬的财富，但其中许多库与 Python 2 时代有遗留联系，并将我们的一流需求作为附加组件而不是核心功能来解决。

我们的设计目标大致是：

• 广泛利用类型提示作为文档和按合同设计的工具，以提供富有表现力、可读性的 API。
• 拥抱 asyncio 作为一等公民。
• 使用易于理解的方法和输入/输出外部 API 界面将状态机实现为普通的旧 Python 对象。呼叫者不需要知道或关心 FSM 细节。
• 提供与大多数开发人员都会接触到的UML 状态机概念框架紧密结合的 API 。
• 避开著名的字符串值，支持 Pythonenum 子类，以促进类型强制、重构和 IDE 完成。
• 使状态机能够以编程方式或声明方式建模。
• 为在状态机类本身中建模数据以及将外部数据传递到转换中提供强大的支持。
• 禁止隐含的“神奇”行为，例如自动创建状态、事件和动作对象或动态定义/调度方法。

最终，我们真的想要一些适合我们现有编码风格和 API 美学的东西。也许政治家也适合你。

过渡生命周期

政治家在转换期间以定义的顺序执行操作。该类 statesman.Transition是一个可调用对象，负责对状态更改进行建模并协调其执行。下表描述了调用转换时执行的操作顺序。

目标	当前状态	注释
StateMachine.guard_transition	source	子类的方法调度。可以取消过渡。
StateMachine.before_transition	source	子类的方法调度。
Event.actions.guard	source	可以取消过渡。
Event.actions.before	source
State.actions.exit	source
StateMachine.on_transition	target	子类的方法调度。
Event.actions.on	target
State.actions.entry	target
Event.actions.after	target
StateMachine.after_transition	target	子类的方法调度。

API 概述

Statesman 被文档字符串和自动化测试广泛覆盖。后续小节提供了 API 的非详尽的面向任务的概述。如果您没有找到您正在寻找的确切内容，请检查文档字符串并查看测试。

初始状态

政治家考虑了一个转换，其中源状态None将描述初始状态转换。在 statesman 中有几种描述初始状态的方法：

import statesman


# Describe the initial state with `statesman.InitialState`
class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    @statesman.event(None, States.starting)
    def start(self) -> None:
        ...


async def _example() -> None:
    # Set at initialization time
    state_machine = StateMachine(state=StateMachine.States.stopping)

    # Enter a state directly
    state_machine = StateMachine()
    await state_machine.enter_state(StateMachine.States.running)

    # Via an event
    state_machine = StateMachine()
    await state_machine.start()

内省状态

每个状态机实例都有一个state属性，该属性是状态机的真实来源。可以将状态与statesman.State对象实例或字符串值进行比较。

import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = 'Terminated.'


async def _example() -> None:
    state_machine = StateMachine(state=StateMachine.States.stopping)
    state_machine.state == StateMachine.States.stopping  # => True
    state_machine.state == "stopping"  # => True
    state_machine.state == StateMachine.States.running  # => False
    state_machine.state == "stopped"  # => False

进入状态

可以通过该statesman.StateMachine.enter_state 方法直接输入状态。状态可以通过名称或stateman.State对象实例来引用。当直接进入状态时，会触发源状态和目标状态之间的转换：

import statesman


class StateMachine(statesman.HistoryMixin, statesman.StateMachine):
    class States(statesman.StateEnum):
        first = "1"
        second = "2"
        third = "3"


async def _example() -> None:
    state_machine = StateMachine()
    await state_machine.enter_state(StateMachine.States.first)
    await state_machine.enter_state(StateMachine.States.second)
    await state_machine.enter_state(StateMachine.States.third)

执行的转换类型是可配置的（见下文）。

请注意，enter_state应该谨慎使用它，因为它启用了可能无法通过事件表达的转换。它的行为也可以被限制和定制。

过渡类型

政治家可以执行三种类型的转换。最常见的类型是机器在两个不同状态之间移动的外部转换。当发生源状态和目标状态相同的转换时，还有另外两种可能的模式：internal和self。

• statesman.Transition.Types.external：状态从一个值更改为另一个值的转换。
• statesman.Transition.Types.internal：源状态和目标状态相同但在转换过程中不退出和重新进入的转换。
• statesman.Transition.Types.self: 源状态和目标状态相同并且在转换过程中退出和重新进入的转换。

转换返回值

为了使外部消费者能够与状态机交互而不暴露其实现细节，Statesman 支持一组灵活的转换返回值。

通过装饰器或编程接口定义事件时，可以配置默认返回类型。statesman.StateMachine.enter_state 当通过orstatesman.StateMachine.trigger_event方法分派转换时，也可以请求显式返回值类型。

转换可以调用任意数量的操作，并且所需的返回值语义因情况而异。考虑您希望向开发人员展示的 API，并仔细考虑在状态机中是否/如何使用None和值。False

可用的返回类型有：

• bool：一个布尔值，指示转换是否成功完成。
• object：转换返回的任意输出值。
• tuple：包含布尔值和输出对象值的元组值。
• list：由转换调用的所有操作返回的结果列表。
• statesman.Transition：过渡对象本身，包含有关过渡的所有详细信息。

请注意，返回类型被配置为类型参数：

import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting'
        running = 'Running'
        stopping = 'Stopping'
        stopped = 'Stopped'

    @statesman.event(None, States.starting, return_type=bool)
    async def start(self) -> int:
        return 31337


async def _example() -> None:
    state_machine = await StateMachine.create()
    bool_result = await state_machine.trigger_event("start")
    int_result = await state_machine.start(return_type=object)
    transition = await state_machine.enter_state(
        StateMachine.States.stopped,
        return_type=statesman.Transition
    )
    print(
        f"Return Values: state_machine={state_machine}\n"
        f"bool_result={bool_result}\n"
        f"int_result={int_result}\n"
        f"transition={transition}"
    )

定义和触发事件

事件通常使用statesman.event装饰器定义。每个事件都有一个源状态和一个目标状态。通常，这些是不同的状态并描述触发外部转换的事件。当源和目标状态相同时，该事件描述了 内部或自我转换（详见上文）。

源和目标状态可以通过字符串名称、枚举成员值或 or 的特殊标记值来statesman.StateEnum.__any__描述 statesman.StateEnum.__active__。sentinel 描述了状态枚举（但不是）__any__的任何成员的源状态。Nonesentinel 动态解析到状态机的__active__ 当前活动状态，并且在您想要定义可以从多个状态触发而无需重复逻辑的自反内部或自我转换的情况下很有用。

事件可以通过名称、方法或调用修饰的事件方法以编程方式触发。

import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        waiting = 'Waiting'
        running = 'Running'
        stopped = 'Stopped'
        aborted = 'Aborted'

    @statesman.event(None, States.waiting)
    async def start(self) -> None:
        ...

    @statesman.event(States.waiting, States.running)
    async def run(self) -> None:
        ...

    @statesman.event(States.running, States.stopped)
    async def stop(self) -> None:
        ...

    @statesman.event(States.__any__, States.aborted)
    async def abort(self) -> None:
        ...

    @statesman.event(
        States.__any__,
        States.__active__,
        type=statesman.Transition.Types.self
    )
    async def check(self) -> None:
        print("Exiting and reentering active state!")


async def _example() -> None:
    state_machine = await StateMachine.create()
    await state_machine.trigger_event("start")
    await state_machine.run()
    await state_machine.trigger_event(state_machine.stop)

状态和事件动作

状态和事件支持任意数量的动作的附加。动作是一个对象，它包装了在状态机生命周期中的指定时刻调用的可调用对象。

状态对象支持以下动作类型：

• statesman.Action.Types.entry：在转换期间进入状态时调用。
• statesman.Action.Types.exit：在转换期间退出状态时调用。

事件动作支持以下动作类型：

• statesman.Action.Types.guard：调用以确定事件是否可以执行。
• statesman.Action.Types.before：在应用事件描述的状态转换之前调用。机器的状态是源状态。
• statesman.Action.Types.on：在应用事件描述的状态转换时调用。机器的状态是目标状态。
• statesman.Action.Types.after：在应用了事件描述的状态转换后调用。机器的状态是目标状态。

可以通过多种方式定义状态和事件操作。statesman.state 和statesman.event装饰器接受以它们支持的动作类型命名的关键字参数。这些参数支持方法对象引用、名称或可调用对象（例如 lambdas）。此外，还有独立的装饰器可以实现声明式的动作定义。

import random
import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    @statesman.event(States.stopped, States.starting)
    async def start(self) -> None:
        ...

    @statesman.enter_state(States.starting)
    async def _announce_start(self) -> None:
        print("enter:starting")

    def _can_run(self) -> bool:
        return False

    @statesman.event(States.starting, States.running, guard=_can_run)
    async def run(self) -> None:
        ...

    @statesman.event(States.running, States.stopping)
    async def stop(self) -> None:
        ...

    @statesman.after_event(stop)
    async def _announce_stop(self) -> None:
        print("after:stop")

    @statesman.event(
        States.stopping,
        States.stopped,
        guard=lambda: random.choice([True, False])
    )
    async def terminate(self) -> None:
        ...

保护回调和操作

Guard 回调和操作的处理方式与其他行为挂钩不同。因为守卫可用于取消/拒绝事件，所以它们按添加到状态机的顺序顺序执行。

在遇到返回False或引发类型异常的失败守卫时AssertionError，状态机会查询嵌套在状态机类中guard_with 的类上配置的行为。Config

可通过以下方式配置三种行为guard_with：

• statesman.Guard.silence（默认）：转换被中止并返回一个空的结果集。
• statesman.Guard.warning：转换被中止，返回一个空的结果集，并记录一个警告。
• statesman.Guard.exception: 转换被中止并RuntimeError引发 a。

import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    class Config:
        guard_with = statesman.Guard.exception

将数据传递给转换

Statesman 旨在建模和管理绑定到状态机当前状态的任意数据。statesman.StateMachine.enter_state 当通过orstatesman.StateMachine.trigger_event方法触发转换时，可以将此类数据作为位置参数和关键字参数提供。

Statesman 利用方法参数列表自省和类型提示来调用带有他们感兴趣的特定参数的回调和操作。通过确保每个回调和方法声明其期望，这有助于更清晰的封装、关注点分离和可维护性。*args和**kwargs可变参数可以根据需要使用。

import statesman


class StateMachine(statesman.StateMachine):
    class States(statesman.StateEnum):
        starting = 'Starting...'
        running = 'Running...'
        stopping = 'Stopping...'
        stopped = statesman.InitialState('Terminated.')

    @statesman.event(States.stopped, States.starting)
    async def start(self, process_name: str) -> None:
        ...

    @statesman.event(States.starting, States.running)
    async def run(self, uid: int, gid: int) -> None:
        ...

    @statesman.event(States.running, States.stopping)
    async def stop(self, *args, all: bool = False, **kwargs) -> None:
        ...


async def _example() -> None:
    state_machine = await StateMachine.create()
    await state_machine.start("servox")
    await state_machine.run(0, 31337)
    await state_machine.stop("one", "two", all=True, this="That")

状态机可继承回调

该类statesman.StateMachine通常是子类。基类提供转换生命周期事件的可继承方法实现。类似的功能可以通过回调来实现，就像通过 State 和 Event 操作一样，只是透视植根于转换而不是 State 或 Event。

有关回调和操作之间的执行顺序和优先级的详细信息，请参阅转换生命周期。

进口 政治家


类 InheritableStateMachine ( statesman . StateMachine <