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

脚本专家

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

一、安装与基础概念

推荐使用uv管理依赖：

2. uv pip install transitions
3. # 如需绘制状态转移图（需安装graphviz）
4. uv pip install transitions[diagrams]

复制代码

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

2. from transitions import Machine
4. states = ['off', 'on']
5. transitions = [
6.     {'trigger': 'turn_on', 'source': 'off', 'dest': 'on'},
7.     {'trigger': 'turn_off', 'source': 'on', 'dest': 'off'}
8. ]
9. machine = Machine(states=states, transitions=transitions, initial='off')
10. print(machine.state)  # off
11. machine.turn_on()
12. print(machine.state)  # on

复制代码

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

二、绑定自定义业务模型

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

2. class LightBulb:
3.     def __init__(self):
4.         self.brightness = 100
6. bulb = LightBulb()
7. states = ['off', 'on', 'broken']
8. machine = Machine(model=bulb, states=states, initial='off')
9. machine.add_transition(trigger='burn_out', source='*', dest='broken')
10. machine.add_transition(trigger='turn_on', source='off', dest='on')
12. print(bulb.state)  # off
13. bulb.turn_on()
14. print(bulb.state)  # on

复制代码

三、条件限制与异常处理

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

2. class Water:
3.     def __init__(self):
4.         self.temperature = 20
5.     def is_hot(self):
6.         return self.temperature >= 100
8. water = Water()
9. machine = Machine(model=water, states=['liquid', 'gas'], initial='liquid', ignore_invalid_triggers=True)
10. machine.add_transition(trigger='evaporate', source='liquid', dest='gas', conditions='is_hot')
12. success = water.evaporate()
13. print(success)  # False
14. water.temperature = 100
15. success = water.evaporate()
16. print(success)  # True

复制代码

四、生命周期回调函数

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

2. class Hero:
3.     def wear_armor(self):
4.         print("[Callback] 穿上战甲！")
5.     def log_success(self):
6.         print("[Callback] 状态成功转换。")
8. hero = Hero()
9. states = ['normal', 'combat']
10. machine = Machine(model=hero, states=states, initial='normal')
11. machine.add_transition(trigger='encounter_enemy', source='normal', dest='combat',
12.                        before='wear_armor', after='log_success')
13. hero.encounter_enemy()
14. # 输出：
15. # [Callback] 穿上战甲！
16. # [Callback] 状态成功转换。

复制代码

五、分层/嵌套状态机

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

2. from transitions.extensions import HierarchicalMachine as Machine
4. states = [
5.     'off',
6.     {'name': 'on', 'children': ['standby', 'working'], 'initial': 'standby'}
7. ]
8. machine = Machine(states=states, initial='off')
9. machine.add_transition('power_on', 'off', 'on')
10. machine.add_transition('activate', 'on_standby', 'on_working')
11. machine.add_transition('power_off', 'on', 'off')
13. print(machine.state)  # off
14. machine.power_on()
15. print(machine.state)  # on_standby
16. machine.activate()
17. print(machine.state)  # on_working

复制代码

六、异步状态机

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

2. import asyncio
3. from transitions.extensions.asyncio import AsyncMachine
5. class AsyncTask:
6.     async def notify_server(self):
7.         print("开始同步")
8.         await asyncio.sleep(0.5)
9.         print("同步完成")
11. task = AsyncTask()
12. machine = AsyncMachine(model=task, states=['pending', 'completed'], initial='pending')
13. machine.add_transition(trigger='complete', source='pending', dest='completed', before='notify_server')
15. async def main():
16.     await task.complete()
17.     print(f"状态: {task.state}")
19. asyncio.run(main())

复制代码

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

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

1. 安装依赖

2. uv pip install sqlalchemy asyncpg transitions

复制代码

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

2. # database.py
3. import asyncio
4. from sqlalchemy import String, Integer
5. from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, DeclarativeBase
6. from sqlalchemy.orm import Mapped, mapped_column
7. from transitions import Machine
9. DATABASE_URL = "postgresql+asyncpg://postgres:password@localhost:5432/mydb"
10. engine = create_async_engine(DATABASE_URL, echo=True)
11. async_session = async_sessionmaker(engine, expire_on_commit=False)
13. class Base(DeclarativeBase):
14.     pass
16. class Order(Base):
17.     __tablename__ = "orders"
18.     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
19.     title: Mapped[str] = mapped_column(String(100))
20.     status: Mapped[str] = mapped_column(String(50), default="created")
22. states = ["created", "paid", "shipped", "completed", "cancelled"]
23. order_machine = Machine(
24.     model=None,
25.     states=states,
26.     initial="created",
27.     model_attribute="status",
28.     ignore_invalid_triggers=True
29. )
30. order_machine.add_transition(trigger="pay", source="created", dest="paid")
31. order_machine.add_transition(trigger="ship", source="paid", dest="shipped")
32. order_machine.add_transition(trigger="complete", source="shipped", dest="completed")
33. order_machine.add_transition(trigger="cancel", source=["created", "paid"], dest="cancelled")

复制代码

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

2. # service.py
3. from database import async_session, Order, order_machine
4. from sqlalchemy import select
6. async def create_new_order(title: str) -> int:
7.     async with async_session() as session:
8.         async with session.begin():
9.             new_order = Order(title=title)
10.             session.add(new_order)
11.             return new_order.id
13. async def process_order_payment(order_id: int):
14.     async with async_session() as session:
15.         async with session.begin():
16.             result = await session.execute(select(Order).where(Order.id == order_id))
17.             order = result.scalar_one_or_none()
18.             if not order:
19.                 print("订单不存在")
20.                 return
21.             order_machine.add_model(order)
22.             success = order.pay()
23.             if success:
24.                 print(f"状态变更为 {order.status}")
25.             else:
26.                 print("转换失败")
27.             order_machine.remove_model(order)

复制代码

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

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

2. from transitions.extensions.asyncio import AsyncMachine
4. class OrderWithCallback(Order):
5.     async def log_status_change(self, event_data):
6.         session = event_data.kwargs.get("session")
7.         if session:
8.             print(f"订单 {self.id} 状态从 {event_data.transition.source} 变更为 {event_data.transition.dest}")
9.             # 可执行额外DB操作，如 session.add(OrderLog(...))
11. async_order_machine = AsyncMachine(
12.     model=None,
13.     states=states,
14.     initial="created",
15.     model_attribute="status"
16. )
17. async_order_machine.add_transition(
18.     trigger="pay",
19.     source="created",
20.     dest="paid",
21.     after="log_status_change"
22. )
24. async def pay_with_callback(order_id: int):
25.     async with async_session() as session:
26.         async with session.begin():
27.             result = await session.execute(select(OrderWithCallback).where(OrderWithCallback.id == order_id))
28.             order = result.scalar_one()
29.             async_order_machine.add_model(order)
30.             await order.pay(session=session)

复制代码

最佳实践总结

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

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

热心网友4

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