"""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": [], } )