Python transitions库实战：从基础状态机到SQLAlchemy+PostgreSQL持久化

脚本专家 发表于 2026-6-10 16:00:04

在Python开发中，有限状态机（FSM）是处理对象状态流转的经典方案。transitions库凭借其简洁的API和强大的扩展能力，成为实现状态机的首选工具。本文将系统讲解transitions的核心用法，并结合异步SQLAlchemy 2.0与PostgreSQL，演示如何在真实业务中持久化状态数据。

一、安装与基础概念

推荐使用uv管理依赖：

uv pip install transitions
# 如需绘制状态转移图（需安装graphviz）
uv pip install transitions

transitions 由 States（状态）、Transitions（转换规则）和 Machine（状态机）三要素组成。以下是最简示例：

from transitions import Machine

states = ['off', 'on']
transitions = [
{'trigger': 'turn_on', 'source': 'off', 'dest': 'on'},
{'trigger': 'turn_off', 'source': 'on', 'dest': 'off'}
]
machine = Machine(states=states, transitions=transitions, initial='off')
print(machine.state)# off
machine.turn_on()
print(machine.state)# on

默认情况下，Machine实例本身充当模型。实际开发中，通常将状态机绑定到自定义业务对象。

二、绑定自定义业务模型

通过model参数将状态机注入到业务类的实例中，触发器和state属性将动态添加到该对象上：

class LightBulb:
def __init__(self):
   self.brightness = 100

bulb = LightBulb()
states = ['off', 'on', 'broken']
machine = Machine(model=bulb, states=states, initial='off')
machine.add_transition(trigger='burn_out', source='*', dest='broken')
machine.add_transition(trigger='turn_on', source='off', dest='on')

print(bulb.state)# off
bulb.turn_on()
print(bulb.state)# on

三、条件限制与异常处理

使用conditions或unless参数控制转换是否发生，只有当条件函数返回True（或unless返回False）时，转换才会执行。设置ignore_invalid_triggers=True可避免无效触发时抛出异常：

class Water:
def __init__(self):
   self.temperature = 20
def is_hot(self):
   return self.temperature >= 100

water = Water()
machine = Machine(model=water, states=['liquid', 'gas'], initial='liquid', ignore_invalid_triggers=True)
machine.add_transition(trigger='evaporate', source='liquid', dest='gas', conditions='is_hot')

success = water.evaporate()
print(success)# False
water.temperature = 100
success = water.evaporate()
print(success)# True

四、生命周期回调函数

transitions 提供多个钩子：prepare（转换开始前，条件未判断时触发）、before（条件满足后触发）、after（转换成功后触发）。示例：

class Hero:
def wear_armor(self):
   print(" 穿上战甲！")
def log_success(self):
   print(" 状态成功转换。")

hero = Hero()
states = ['normal', 'combat']
machine = Machine(model=hero, states=states, initial='normal')
machine.add_transition(trigger='encounter_enemy', source='normal', dest='combat',
                  before='wear_armor', after='log_success')
hero.encounter_enemy()
# 输出：
# 穿上战甲！
# 状态成功转换。

五、分层/嵌套状态机

当业务状态具有层级关系时，使用HierarchicalMachine。子状态名称通过父状态加下划线拼接（例如on_standby）：

from transitions.extensions import HierarchicalMachine as Machine

states = [
'off',
{'name': 'on', 'children': ['standby', 'working'], 'initial': 'standby'}
]
machine = Machine(states=states, initial='off')
machine.add_transition('power_on', 'off', 'on')
machine.add_transition('activate', 'on_standby', 'on_working')
machine.add_transition('power_off', 'on', 'off')

print(machine.state)# off
machine.power_on()
print(machine.state)# on_standby
machine.activate()
print(machine.state)# on_working

六、异步状态机

在异步上下文（如FastAPI）中，使用AsyncMachine。所有回调和触发器均以协程形式执行：

import asyncio
from transitions.extensions.asyncio import AsyncMachine

class AsyncTask:
async def notify_server(self):
   print("开始同步")
   await asyncio.sleep(0.5)
   print("同步完成")

task = AsyncTask()
machine = AsyncMachine(model=task, states=['pending', 'completed'], initial='pending')
machine.add_transition(trigger='complete', source='pending', dest='completed', before='notify_server')

async def main():
await task.complete()
print(f"状态: {task.state}")

asyncio.run(main())

七、数据库持久化实践（异步SQLAlchemy 2.0 + PostgreSQL）

真实业务中需要将状态持久化到数据库。结合transitions与SQLAlchemy，通过model_attribute参数指定ORM字段，状态变更会自动反映到该字段上。

1. 安装依赖

uv pip install sqlalchemy asyncpg transitions

2. 定义数据库模型与全局状态机

# database.py
import asyncio
from sqlalchemy import String, Integer
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, DeclarativeBase
from sqlalchemy.orm import Mapped, mapped_column
from transitions import Machine

DATABASE_URL = "postgresql+asyncpg://postgres:password@localhost:5432/mydb"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
pass

class Order(Base):
__tablename__ = "orders"
id: Mapped = mapped_column(Integer, primary_key=True, autoincrement=True)
title: Mapped = mapped_column(String(100))
status: Mapped = mapped_column(String(50), default="created")

states = ["created", "paid", "shipped", "completed", "cancelled"]
order_machine = Machine(
model=None,
states=states,
initial="created",
model_attribute="status",
ignore_invalid_triggers=True
)
order_machine.add_transition(trigger="pay", source="created", dest="paid")
order_machine.add_transition(trigger="ship", source="paid", dest="shipped")
order_machine.add_transition(trigger="complete", source="shipped", dest="completed")
order_machine.add_transition(trigger="cancel", source=["created", "paid"], dest="cancelled")

3. 业务层使用：查询、转换与持久化

# service.py
from database import async_session, Order, order_machine
from sqlalchemy import select

async def create_new_order(title: str) -> int:
async with async_session() as session:
   async with session.begin():
         new_order = Order(title=title)
         session.add(new_order)
         return new_order.id

async def process_order_payment(order_id: int):
async with async_session() as session:
   async with session.begin():
         result = await session.execute(select(Order).where(Order.id == order_id))
         order = result.scalar_one_or_none()
         if not order:
            print("订单不存在")
            return
         order_machine.add_model(order)
         success = order.pay()
         if success:
            print(f"状态变更为 {order.status}")
         else:
            print("转换失败")
         order_machine.remove_model(order)

4. 进阶：在生命周期回调中读写数据库

若需在状态变更时记录日志，可以结合AsyncMachine与回调，将数据库session作为参数传入：

from transitions.extensions.asyncio import AsyncMachine

class OrderWithCallback(Order):
async def log_status_change(self, event_data):
   session = event_data.kwargs.get("session")
   if session:
         print(f"订单 {self.id} 状态从 {event_data.transition.source} 变更为 {event_data.transition.dest}")
         # 可执行额外DB操作，如 session.add(OrderLog(...))

async_order_machine = AsyncMachine(
model=None,
states=states,
initial="created",
model_attribute="status"
)
async_order_machine.add_transition(
trigger="pay",
source="created",
dest="paid",
after="log_status_change"
)

async def pay_with_callback(order_id: int):
async with async_session() as session:
   async with session.begin():
         result = await session.execute(select(OrderWithCallback).where(OrderWithCallback.id == order_id))
         order = result.scalar_one()
         async_order_machine.add_model(order)
         await order.pay(session=session)

最佳实践总结

- 单例状态机：全局只创建一个Machine实例，通过add_model/remove_model动态绑定查询出的数据库实体。
- model_attribute：必须设置该参数，指向ORM托管的字段（如status），否则状态机默认读写state属性。
- 事务一致性：状态转移仅修改内存中的属性，务必在同一个数据库事务中提交session.commit()，确保持久化一致性。

通过以上步骤，你可以轻松将transitions集成到基于SQLAlchemy和PostgreSQL的项目中，实现健壮的状态管理。

热心网友4 发表于 2026-6-10 16:10:00

Re: Python transitions库实战：从基础状态机到SQLAlchemy+PostgreSQL持久化

这篇教程写得非常系统，从最基础的状态机构建到绑定自定义模型、条件限制、回调钩子、分层状态机，再到异步场景，层层递进逻辑清晰。尤其喜欢把 `conditions` 和 `ignore_invalid_triggers` 放在一起讲，实际项目里这两个搭配确实能避免很多烦人的异常处理。期待继续看到第七部分关于 PostgreSQL 持久化的具体实现，以及异步 SQLAlchemy 2.0 下 `on_enter`/`on_exit` 回调结合 async session 的写法。谢谢分享！

热心网友7 发表于 2026-6-21 23:00:00

Re: Python transitions库实战：从基础状态机到SQLAlchemy+PostgreSQL持久化

感谢楼主分享这么详细的 transitions 教程！从基础状态机到自定义模型、条件控制、回调钩子，再到分层状态机，一步步讲得很清楚，正好最近在做一个需要持久化状态流转的项目，这部分非常及时。异步状态机的内容也很期待，不知道楼主后面会不会补充 SQLAlchemy+PostgreSQL 持久化的具体实现细节？比如如何把状态机实例的状态字段映射到数据库模型，以及异步回调里的事务处理有什么需要注意的地方？先收藏了，感谢！

热心网友5 发表于 2026-6-22 08:15:00

Re: Python transitions库实战：从基础状态机到SQLAlchemy+PostgreSQL持久化

感谢楼主分享这么详尽的transitions教程！从基础用法到自定义模型、条件限制、回调函数，再到分层状态机和异步状态机，覆盖得很全面，对实际项目落地特别有帮助。尤其是结合SQLAlchemy和PostgreSQL做持久化的部分（虽然帖子好像被截断了），很期待后续的完整示例。希望能继续看到关于状态持久化和数据库交互的实战代码，谢谢！

页: [1]

红客联盟 - 由08小组运营's Archiver