153 lines
4.6 KiB
Python
153 lines
4.6 KiB
Python
"""通用状态机服务。
|
||
|
||
设计：
|
||
- 单据状态变化走 StateMachine.transition()，不在 model.save() 里散写
|
||
- 支持 guard（条件校验）、action（副作用：扣库存/生成应收）、hook（插件/AI Agent/Webhook）
|
||
- 可被 view / service / task 统一调用
|
||
- 默认包 transaction.atomic()；use_transaction=False 用于纯内存测试
|
||
|
||
用法：
|
||
sm = StateMachine(
|
||
states=("draft", "confirmed", "posted", "closed", "cancelled"),
|
||
initial="draft",
|
||
transitions={
|
||
"draft": {"confirm": "confirmed", "cancel": "cancelled"},
|
||
"confirmed": {"post": "posted", "cancel": "cancelled"},
|
||
"posted": {"close": "closed"},
|
||
},
|
||
guards={
|
||
("draft", "confirmed", "confirm"): require_draft,
|
||
},
|
||
actions={
|
||
("draft", "confirmed", "confirm"): on_confirm,
|
||
},
|
||
hooks={"post": [notify_webhook, log_audit]},
|
||
)
|
||
sm.transition(instance, "confirm", user=request.user)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from typing import Iterable, Optional
|
||
|
||
from django.db import transaction
|
||
|
||
|
||
@dataclass
|
||
class TransitionContext:
|
||
instance: object
|
||
from_state: str
|
||
to_state: str
|
||
event: str
|
||
user: Optional[object] = None
|
||
extra: dict = field(default_factory=dict)
|
||
|
||
|
||
class StateTransitionError(Exception):
|
||
"""状态机非法迁移。"""
|
||
|
||
pass
|
||
|
||
|
||
class StateMachine:
|
||
def __init__(
|
||
self,
|
||
states: Iterable[str],
|
||
initial: str,
|
||
transitions: dict,
|
||
guards: dict | None = None,
|
||
actions: dict | None = None,
|
||
hooks: dict | None = None,
|
||
) -> None:
|
||
self.states = tuple(states)
|
||
self.initial = initial
|
||
self.transitions = {s: dict(es) for s, es in transitions.items()}
|
||
self.guards = guards or {}
|
||
self.actions = actions or {}
|
||
self.hooks = hooks or {}
|
||
if initial not in self.states:
|
||
raise ValueError(f"initial state {initial!r} not in states")
|
||
|
||
# --- 公共 API ---
|
||
|
||
def get_next_states(self, current: str) -> list[str]:
|
||
return list(self.transitions.get(current, {}).values())
|
||
|
||
def get_events(self, current: str) -> list[str]:
|
||
return list(self.transitions.get(current, {}).keys())
|
||
|
||
def transition(
|
||
self,
|
||
instance,
|
||
event: str,
|
||
*,
|
||
user=None,
|
||
extra: dict | None = None,
|
||
use_transaction: bool = True,
|
||
) -> TransitionContext:
|
||
"""执行一次状态迁移。
|
||
|
||
1. guard 校验
|
||
2. 改 instance.state
|
||
3. action 执行
|
||
4. hook 触发
|
||
5. 返回 TransitionContext
|
||
|
||
use_transaction=False 时不包 transaction.atomic()（用于纯内存测试）。
|
||
"""
|
||
if use_transaction:
|
||
with transaction.atomic():
|
||
return self._run(instance, event, user=user, extra=extra)
|
||
return self._run(instance, event, user=user, extra=extra)
|
||
|
||
def _run(self, instance, event, *, user, extra):
|
||
extra = dict(extra or {})
|
||
from_state = getattr(instance, "state", None)
|
||
if from_state not in self.transitions:
|
||
raise StateTransitionError(
|
||
f"current state {from_state!r} has no outgoing transitions"
|
||
)
|
||
if event not in self.transitions[from_state]:
|
||
raise StateTransitionError(
|
||
f"event {event!r} not allowed from state {from_state!r}"
|
||
)
|
||
|
||
to_state = self.transitions[from_state][event]
|
||
ctx = TransitionContext(
|
||
instance=instance,
|
||
from_state=from_state,
|
||
to_state=to_state,
|
||
event=event,
|
||
user=user,
|
||
extra=extra,
|
||
)
|
||
|
||
guard = self.guards.get((from_state, to_state, event))
|
||
if guard:
|
||
try:
|
||
ok = guard(ctx)
|
||
except Exception:
|
||
raise
|
||
else:
|
||
if ok is False:
|
||
raise StateTransitionError(
|
||
f"guard rejected transition {from_state} -[ {event} ]-> {to_state}"
|
||
)
|
||
|
||
instance.state = to_state
|
||
# save 由调用方负责（Django model.save() 或测试 stub）
|
||
|
||
action = self.actions.get((from_state, to_state, event))
|
||
if action:
|
||
action(ctx)
|
||
|
||
for hook in self.hooks.get(event, []):
|
||
hook(ctx)
|
||
|
||
return ctx
|
||
|
||
def can_transition(self, instance, event: str) -> bool:
|
||
from_state = getattr(instance, "state", None)
|
||
return event in self.transitions.get(from_state, {})
|