67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
|
"""Generic finite state machine for domain entity status transitions.
|
||
|
|
|
||
|
|
Defines allowed state transitions for Contact and Workflow entities.
|
||
|
|
Usage in Commands:
|
||
|
|
|
||
|
|
from app.core.state_machine import contact_state_machine
|
||
|
|
contact_state_machine.transition(contact.status, "qualified")
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
|
||
|
|
class StateMachineError(Exception):
|
||
|
|
"""Raised when an invalid state transition is attempted."""
|
||
|
|
|
||
|
|
|
||
|
|
class StateMachine:
|
||
|
|
"""Finite state machine that validates and executes state transitions.
|
||
|
|
|
||
|
|
Attributes:
|
||
|
|
transitions: Mapping from a state to the list of states it can transition to.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, transitions: dict[str, list[str]]) -> None:
|
||
|
|
self.transitions: dict[str, list[str]] = transitions
|
||
|
|
|
||
|
|
def can_transition(self, current: str, target: str) -> bool:
|
||
|
|
"""Return True if transitioning from *current* to *target* is allowed."""
|
||
|
|
allowed = self.transitions.get(current, [])
|
||
|
|
return target in allowed
|
||
|
|
|
||
|
|
def transition(self, current: str, target: str) -> str:
|
||
|
|
"""Validate and return the new state.
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
StateMachineError: if the transition is not allowed.
|
||
|
|
"""
|
||
|
|
if not self.can_transition(current, target):
|
||
|
|
raise StateMachineError(
|
||
|
|
f"Invalid state transition: '{current}' -> '{target}'. "
|
||
|
|
f"Allowed targets from '{current}': {self.transitions.get(current, [])}"
|
||
|
|
)
|
||
|
|
return target
|
||
|
|
|
||
|
|
|
||
|
|
# ── Contact lifecycle: lead → qualified → customer → inactive ──
|
||
|
|
# Allows skipping 'qualified' and reactivation from inactive.
|
||
|
|
contact_state_machine = StateMachine(
|
||
|
|
transitions={
|
||
|
|
"lead": ["qualified", "customer", "inactive"],
|
||
|
|
"qualified": ["customer", "lead", "inactive"],
|
||
|
|
"customer": ["inactive"],
|
||
|
|
"inactive": ["lead"],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
# ── Workflow lifecycle: draft → active → paused → completed → cancelled ──
|
||
|
|
workflow_state_machine = StateMachine(
|
||
|
|
transitions={
|
||
|
|
"draft": ["active", "cancelled"],
|
||
|
|
"active": ["paused", "completed", "cancelled"],
|
||
|
|
"paused": ["active", "completed", "cancelled"],
|
||
|
|
"completed": [],
|
||
|
|
"cancelled": [],
|
||
|
|
}
|
||
|
|
)
|