# 基礎(chǔ)安裝
uv pip install transitions
# 如果需要支持繪制狀態(tài)轉(zhuǎn)移圖 (需要系統(tǒng)已安裝 graphviz)
uv pip install transitions[diagrams]

from transitions import Machine

# 1. 定義狀態(tài)列表
states = ['off', 'on']

# 2. 定義狀態(tài)轉(zhuǎn)換規(guī)則
# trigger: 觸發(fā)動作的方法名
# source: 轉(zhuǎn)換前的狀態(tài)
# dest: 轉(zhuǎn)換后的目標(biāo)狀態(tài)
transitions = [
    {'trigger': 'turn_on', 'source': 'off', 'dest': 'on'},
    {'trigger': 'turn_off', 'source': 'on', 'dest': 'off'}
]

# 3. 初始化狀態(tài)機
# 默認(rèn)情況下，如果不傳 model，Machine 實例本身將作為 Model
machine = Machine(states=states, transitions=transitions, initial='off')

# 4. 使用狀態(tài)機
print(machine.state)  # 輸出: off
machine.turn_on()     # 觸發(fā) 'turn_on'
print(machine.state)  # 輸出: on

from transitions import Machine

# 業(yè)務(wù)類
class LightBulb:
    def __init__(self):
        self.brightness = 100

bulb = LightBulb()
states = ['off', 'on', 'broken']

# 將狀態(tài)機綁定到 bulb 實例
machine = Machine(model=bulb, states=states, initial='off')

# 動態(tài)添加轉(zhuǎn)換規(guī)則
# '*' 代表可以從任何狀態(tài)轉(zhuǎn)換到 'broken'
machine.add_transition(trigger='burn_out', source='*', dest='broken')
machine.add_transition(trigger='turn_on', source='off', dest='on')

# 此時 bulb 對象擁有了 state 屬性和觸發(fā)器方法
print(bulb.state)  # 輸出: off
bulb.turn_on()
print(bulb.state)  # 輸出: on
bulb.burn_out()
print(bulb.state)  # 輸出: broken

from transitions import Machine, MachineError

class Water:
    def __init__(self):
        self.temperature = 20

    def is_hot(self):
        return self.temperature >= 100

water = Water()
states = ['liquid', 'gas']
# ignore_invalid_triggers=True 可以防止在無效狀態(tài)下調(diào)用觸發(fā)器拋出異常，而是返回 False
machine = Machine(model=water, states=states, initial='liquid', ignore_invalid_triggers=True)

machine.add_transition(
    trigger='evaporate', 
    source='liquid', 
    dest='gas', 
    conditions='is_hot'  # 綁定條件檢查方法
)

# 嘗試在 20 度時蒸發(fā)
success = water.evaporate()
print(success)      # 輸出: False (轉(zhuǎn)換失敗)
print(water.state)  # 輸出: liquid

# 加熱后再次嘗試
water.temperature = 100
success = water.evaporate()
print(success)      # 輸出: True (轉(zhuǎn)換成功)
print(water.state)  # 輸出: gas

class Hero:
    def wear_armor(self):
        print("[Callback] 穿上戰(zhàn)甲！")

    def log_success(self):
        print("[Callback] 狀態(tài)成功轉(zhuǎn)換。")

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()
# 控制臺輸出:
# [Callback] 穿上戰(zhàn)甲！
# [Callback] 狀態(tài)成功轉(zhuǎn)換。

from transitions.extensions import HierarchicalMachine as Machine

# 定義嵌套狀態(tài)結(jié)構(gòu)
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')
# 從 on 下的任何子狀態(tài)均可 power_off 轉(zhuǎn)換到 off
machine.add_transition('power_off', 'on', 'off')

print(machine.state)  # 輸出: off
machine.power_on()
print(machine.state)  # 輸出: on_standby (進(jìn)入 on 狀態(tài)的默認(rèn)初始子狀態(tài))
machine.activate()
print(machine.state)  # 輸出: on_working

import asyncio
from transitions.extensions.asyncio import AsyncMachine

class AsyncTask:
    async def notify_server(self):
        print("開始向服務(wù)器同步狀態(tài)...")
        await asyncio.sleep(0.5)
        print("同步完成。")

task = AsyncTask()
states = ['pending', 'completed']
machine = AsyncMachine(model=task, states=states, initial='pending')

machine.add_transition(
    trigger='complete', 
    source='pending', 
    dest='completed',
    before='notify_server'
)

async def main():
    # 異步觸發(fā)器必須使用 await
    await task.complete()
    print(f"當(dāng)前狀態(tài): {task.state}")

asyncio.run(main())

uv pip install sqlalchemy asyncpg transitions

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

# 1. 初始化數(shù)據(jù)庫連接 (請?zhí)鎿Q為您的 PostgreSQL 連接串)
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

# 2. 定義 SQLAlchemy 訂單模型
class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(100))
    # 狀態(tài)字段，對應(yīng)狀態(tài)機中的狀態(tài)
    status: Mapped[str] = mapped_column(String(50), default="created")

# 3. 初始化全局狀態(tài)機 (不綁定具體 model)
# model_attribute='status' 關(guān)鍵參數(shù)：指示狀態(tài)機讀寫 model 的 status 屬性而不是默認(rèn)的 state
states = ["created", "paid", "shipped", "completed", "cancelled"]
order_machine = Machine(
    model=None,  # 初始不綁定任何 model
    states=states, 
    initial="created", 
    model_attribute="status",
    ignore_invalid_triggers=True
)

# 4. 定義轉(zhuǎn)換規(guī)則
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")

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

async def create_new_order(title: str) -> int:
    """創(chuàng)建新訂單，初始狀態(tài)為 'created'"""
    async with async_session() as session:
        async with session.begin():
            new_order = Order(title=title)
            session.add(new_order)
        # commit 在 begin() 塊結(jié)束時自動發(fā)生
        return new_order.id

async def process_order_payment(order_id: int):
    """處理訂單支付：狀態(tài)從 'created' -> 'paid'"""
    async with async_session() as session:
        async with session.begin():
            # 1. 從 PostgreSQL 查詢訂單
            result = await session.execute(select(Order).where(Order.id == order_id))
            order = result.scalar_one_or_none()
            
            if not order:
                print("訂單不存在")
                return

            # 2. 將此訂單實例注冊到全局狀態(tài)機中
            # transitions 內(nèi)部使用弱引用(weakref)管理 model，不會造成內(nèi)存泄漏
            order_machine.add_model(order)

            # 3. 觸發(fā)轉(zhuǎn)換 (會自動修改 order.status 屬性)
            print(f"當(dāng)前數(shù)據(jù)庫狀態(tài): {order.status}")  # created
            success = order.pay()  # 觸發(fā) 'pay'
            
            if success:
                print(f"內(nèi)存狀態(tài)已變更為: {order.status}")  # paid
                # 4. 離開 begin() 塊時，SQLAlchemy 會自動發(fā)出 UPDATE 語句持久化到 PostgreSQL
            else:
                print("狀態(tài)轉(zhuǎn)換失敗，不滿足轉(zhuǎn)換條件")

            # 5. 轉(zhuǎn)換結(jié)束，從狀態(tài)機中注銷 model (可選，弱引用會自動回收，但手動移除更干凈)
            order_machine.remove_model(order)

from transitions.extensions.asyncio import AsyncMachine
from sqlalchemy.ext.asyncio import AsyncSession

# 1. 聲明包含 DB 回調(diào)邏輯的 Model 基類或 Mixin
class OrderWithCallback(Order):
    # 此方法將作為 transitions 的 before/after 回調(diào)運行
    async def log_status_change(self, event_data):
        # 從觸發(fā)參數(shù)中獲取當(dāng)前的 db session
        session: AsyncSession = event_data.kwargs.get("session")
        if session:
            print(f"正在向日志表寫入：訂單 {self.id} 狀態(tài)從 {event_data.transition.source} 變更為 {event_data.transition.dest}")
            # 執(zhí)行額外的 DB 插入操作，例如: session.add(OrderLog(...))

# 2. 使用 AsyncMachine
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"  # 綁定回調(diào)
)

# 3. 業(yè)務(wù)調(diào)用
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)
            # 傳入 session 供回調(diào)函數(shù)使用
            await order.pay(session=session)