State Machine

Core FSM class: load config from YAML or dict, process transitions, and run guards and actions.

StateMachine

StateMachine(
    states: dict[str, State],
    transitions: list[Transition],
    meta: dict[str, Any] | None = None,
    error_policy: ErrorPolicy | None = None,
)

Config-driven finite state machine definition and factory.

The StateMachine is stateless -- it holds the FSM definition (states, transitions, guards, actions) but no per-entity state. Use :meth:create to get a stateful :class:EntitySession.

Config is the single source of truth

• States, transitions, guards, and actions are declared in YAML/dict.
• Python implementations are bound via @machine.guard("name") and @machine.action("name") decorators.

Source code in src/pystator/machine.py

def __init__(
    self,
    states: dict[str, State],
    transitions: list[Transition],
    meta: dict[str, Any] | None = None,
    error_policy: ErrorPolicy | None = None,
) -> None:
    self._states = states
    self._transitions = transitions
    self._meta = meta or {}
    self._error_policy = error_policy or ErrorPolicy(
        strict_mode=self._meta.get("strict_mode", True)
    )

    # Internal engine
    self._engine = Engine(states, transitions, self._meta, self._error_policy)

    # Registries (populated via decorators or bind methods)
    self._guard_registry = GuardRegistry()
    self._action_registry = ActionRegistry()

    # Callback registries for on_enter/on_exit/on_transition decorators
    self._on_enter_callbacks: dict[str, list[Callable]] = {}
    self._on_exit_callbacks: dict[str, list[Callable]] = {}
    self._on_transition_callbacks: dict[tuple[str, str], list[Callable]] = {}

    # Bind guard checker to engine
    evaluator = GuardEvaluator(
        self._guard_registry, strict=self._error_policy.strict_mode
    )
    self._engine.guard_checker._registry = self._guard_registry
    self._engine.guard_checker._evaluator = evaluator
    self._engine.guard_checker.strict = self._error_policy.strict_mode

name property

name: str

Machine name from metadata.

version property

version: str

Machine version from metadata.

states property

states: dict[str, State]

All states (read-only copy).

transitions property

transitions: list[Transition]

All transitions (read-only copy).

from_yaml classmethod

from_yaml(
    path: str | Path,
    validate: bool = True,
    variables: dict[str, str] | None = None,
) -> StateMachine

Create a StateMachine from a YAML configuration file.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the YAML configuration file.	required
validate	bool	If True, validate config against schema.	True
variables	dict[str, str] | None	Optional variables for ${VAR} substitution.	None

Returns:

Type	Description
StateMachine	Configured StateMachine instance.

Source code in src/pystator/machine.py

@classmethod
def from_yaml(
    cls,
    path: str | Path,
    validate: bool = True,
    variables: dict[str, str] | None = None,
) -> StateMachine:
    """Create a StateMachine from a YAML configuration file.

    Args:
        path: Path to the YAML configuration file.
        validate: If True, validate config against schema.
        variables: Optional variables for ``${VAR}`` substitution.

    Returns:
        Configured StateMachine instance.
    """
    loader = ConfigLoader(validate=validate, variables=variables)
    config = loader.load(path)
    return cls.from_dict(config)

from_dict classmethod

from_dict(config: dict[str, Any]) -> StateMachine

Create a StateMachine from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Configuration dictionary matching the schema.	required

Returns:

Type	Description
StateMachine	Configured StateMachine instance.

Source code in src/pystator/machine.py

@classmethod
def from_dict(cls, config: dict[str, Any]) -> StateMachine:
    """Create a StateMachine from a configuration dictionary.

    Args:
        config: Configuration dictionary matching the schema.

    Returns:
        Configured StateMachine instance.
    """
    loader = ConfigLoader(validate=False)
    states, transitions, meta = loader.parse(config)

    ep_raw = config.get("error_policy")
    if ep_raw:
        error_policy = ErrorPolicy(
            default_fallback=ep_raw.get("default_fallback"),
            retry_attempts=ep_raw.get("retry_attempts", 0),
            strict_mode=meta.get("strict_mode", True),
        )
    else:
        error_policy = ErrorPolicy(strict_mode=meta.get("strict_mode", True))

    return cls(states, transitions, meta, error_policy)

guard

guard(name: str) -> Callable[[AnyGuardFunc], AnyGuardFunc]

Decorator to bind a guard implementation to its config name.

Example

@machine.guard("is_valid") ... def is_valid(ctx): ... return ctx.get("valid", False)

Source code in src/pystator/machine.py

def guard(self, name: str) -> Callable[[AnyGuardFunc], AnyGuardFunc]:
    """Decorator to bind a guard implementation to its config name.

    Example:
        >>> @machine.guard("is_valid")
        ... def is_valid(ctx):
        ...     return ctx.get("valid", False)
    """

    def decorator(func: AnyGuardFunc) -> AnyGuardFunc:
        self._guard_registry.register(name, func)
        return func

    return decorator

action

action(
    name: str,
) -> Callable[[AnyActionFunc], AnyActionFunc]

Decorator to bind an action implementation to its config name.

Example

@machine.action("notify") ... def notify(ctx): ... send_email(ctx["email"], "State changed")

Source code in src/pystator/machine.py

def action(self, name: str) -> Callable[[AnyActionFunc], AnyActionFunc]:
    """Decorator to bind an action implementation to its config name.

    Example:
        >>> @machine.action("notify")
        ... def notify(ctx):
        ...     send_email(ctx["email"], "State changed")
    """

    def decorator(func: AnyActionFunc) -> AnyActionFunc:
        self._action_registry.register(name, func)
        return func

    return decorator

on_enter

on_enter(state: str) -> Callable[[F], F]

Decorator to register a callback when state is entered.

Example

@machine.on_enter("shipped") ... def notify_customer(ctx): ... send_email(ctx["email"], "Your order shipped!")

Source code in src/pystator/machine.py

def on_enter(self, state: str) -> Callable[[F], F]:
    """Decorator to register a callback when *state* is entered.

    Example:
        >>> @machine.on_enter("shipped")
        ... def notify_customer(ctx):
        ...     send_email(ctx["email"], "Your order shipped!")
    """

    def decorator(func: F) -> F:
        self._on_enter_callbacks.setdefault(state, []).append(func)
        return func

    return decorator

on_exit

on_exit(state: str) -> Callable[[F], F]

Decorator to register a callback when state is exited.

Example

@machine.on_exit("pending") ... def log_departure(ctx): ... logger.info("Left pending: %s", ctx.get("order_id"))

Source code in src/pystator/machine.py

def on_exit(self, state: str) -> Callable[[F], F]:
    """Decorator to register a callback when *state* is exited.

    Example:
        >>> @machine.on_exit("pending")
        ... def log_departure(ctx):
        ...     logger.info("Left pending: %s", ctx.get("order_id"))
    """

    def decorator(func: F) -> F:
        self._on_exit_callbacks.setdefault(state, []).append(func)
        return func

    return decorator

on_transition

on_transition(source: str, target: str) -> Callable[[F], F]

Decorator to register a callback for a specific transition.

Example

@machine.on_transition("pending", "confirmed") ... def celebrate(ctx): ... logger.info("Order confirmed!")

Source code in src/pystator/machine.py

def on_transition(self, source: str, target: str) -> Callable[[F], F]:
    """Decorator to register a callback for a specific transition.

    Example:
        >>> @machine.on_transition("pending", "confirmed")
        ... def celebrate(ctx):
        ...     logger.info("Order confirmed!")
    """

    def decorator(func: F) -> F:
        key = (source, target)
        self._on_transition_callbacks.setdefault(key, []).append(func)
        return func

    return decorator

bind_guards

bind_guards(registry: GuardRegistry) -> StateMachine

Bind an external guard registry.

Replaces the internal guard registry with the provided one.

Returns:

Type	Description
StateMachine	Self for method chaining.

Source code in src/pystator/machine.py

def bind_guards(self, registry: GuardRegistry) -> StateMachine:
    """Bind an external guard registry.

    Replaces the internal guard registry with the provided one.

    Returns:
        Self for method chaining.
    """
    self._guard_registry = registry
    evaluator = GuardEvaluator(registry, strict=self._error_policy.strict_mode)
    self._engine.guard_checker._registry = registry
    self._engine.guard_checker._evaluator = evaluator
    return self

bind_actions

bind_actions(registry: ActionRegistry) -> StateMachine

Bind an external action registry.

Returns:

Type	Description
StateMachine	Self for method chaining.

Source code in src/pystator/machine.py

def bind_actions(self, registry: ActionRegistry) -> StateMachine:
    """Bind an external action registry.

    Returns:
        Self for method chaining.
    """
    self._action_registry = registry
    return self

process

process(
    current_state: str,
    event: str | Event,
    context: dict[str, Any] | None = None,
) -> TransitionResult

Process an event and compute the transition result.

Pure computation -- no side effects, no state mutation.

Parameters:

Name	Type	Description	Default
current_state	str	The current state name.	required
event	str | Event	Event trigger string or Event object.	required
context	dict[str, Any] | None	Optional context dict for guards.	None

Returns:

Type	Description
TransitionResult	TransitionResult describing the computed transition.

Source code in src/pystator/machine.py

def process(
    self,
    current_state: str,
    event: str | Event,
    context: dict[str, Any] | None = None,
) -> TransitionResult:
    """Process an event and compute the transition result.

    Pure computation -- no side effects, no state mutation.

    Args:
        current_state: The current state name.
        event: Event trigger string or Event object.
        context: Optional context dict for guards.

    Returns:
        TransitionResult describing the computed transition.
    """
    return self._engine.process(current_state, event, context)

aprocess async

aprocess(
    current_state: str,
    event: str | Event,
    context: dict[str, Any] | None = None,
) -> TransitionResult

Async version of :meth:process (supports async guards).

Source code in src/pystator/machine.py

async def aprocess(
    self,
    current_state: str,
    event: str | Event,
    context: dict[str, Any] | None = None,
) -> TransitionResult:
    """Async version of :meth:`process` (supports async guards)."""
    return await self._engine.aprocess(current_state, event, context)

create

create(
    *,
    entity_id: str | None = None,
    context: dict[str, Any] | None = None,
    initial_state: str | None = None
) -> EntitySession

Create a stateful instance of this machine.

Parameters:

Name	Type	Description	Default
entity_id	str | None	Optional identifier (auto-generated UUID if omitted).	None
context	dict[str, Any] | None	Initial context dict for the instance.	None
initial_state	str | None	Override the initial state (defaults to config initial).	None

Returns:

Type	Description
EntitySession	A new :class:EntitySession.

Source code in src/pystator/machine.py

def create(
    self,
    *,
    entity_id: str | None = None,
    context: dict[str, Any] | None = None,
    initial_state: str | None = None,
) -> EntitySession:
    """Create a stateful instance of this machine.

    Args:
        entity_id: Optional identifier (auto-generated UUID if omitted).
        context: Initial context dict for the instance.
        initial_state: Override the initial state (defaults to config initial).

    Returns:
        A new :class:`EntitySession`.
    """
    # Lazy import to avoid circular dependency
    from pystator.instance import EntitySession

    return EntitySession(
        machine=self,
        entity_id=entity_id,
        context=context,
        initial_state=initial_state,
    )

orchestrate

orchestrate(
    store: Any,
    *,
    context_validator: Any = None,
    scheduler: Any = None,
    use_initial_state_when_missing: bool = True,
    invoke_adapter: Any = None,
    hooks: list[Any] | None = None,
    audit_store: Any = None
) -> Any

Create an Orchestrator with this machine's registries.

Convenience factory that eliminates the need to pass guard and action registries separately — they are taken from this machine.

Parameters:

Name	Type	Description	Default
store	Any	Sync or async state store.	required
context_validator	Any	Optional pre-transition context validator.	None
scheduler	Any	Optional scheduler for delayed transitions.	None
use_initial_state_when_missing	bool	Use initial state for new entities.	True
invoke_adapter	Any	Optional adapter for invoked services.	None
hooks	list[Any] | None	Optional list of TransitionHook implementations.	None
audit_store	Any	Optional AuditStore for transition audit logging.	None

Returns:

Name	Type	Description
Configured	Any	class:Orchestrator.

Source code in src/pystator/machine.py

def orchestrate(
    self,
    store: Any,
    *,
    context_validator: Any = None,
    scheduler: Any = None,
    use_initial_state_when_missing: bool = True,
    invoke_adapter: Any = None,
    hooks: list[Any] | None = None,
    audit_store: Any = None,
) -> Any:
    """Create an Orchestrator with this machine's registries.

    Convenience factory that eliminates the need to pass guard and
    action registries separately — they are taken from this machine.

    Args:
        store: Sync or async state store.
        context_validator: Optional pre-transition context validator.
        scheduler: Optional scheduler for delayed transitions.
        use_initial_state_when_missing: Use initial state for new entities.
        invoke_adapter: Optional adapter for invoked services.
        hooks: Optional list of TransitionHook implementations.
        audit_store: Optional AuditStore for transition audit logging.

    Returns:
        Configured :class:`Orchestrator`.
    """
    from pystator.orchestrator import Orchestrator

    return Orchestrator(
        machine=self,
        state_store=store,
        context_validator=context_validator,
        scheduler=scheduler,
        use_initial_state_when_missing=use_initial_state_when_missing,
        invoke_adapter=invoke_adapter,
        hooks=hooks,
        audit_store=audit_store,
    )

get_state

get_state(name: str) -> State

Get a state definition by name.

Source code in src/pystator/machine.py

def get_state(self, name: str) -> State:
    """Get a state definition by name."""
    from pystator.errors import UndefinedStateError

    if name not in self._states:
        raise UndefinedStateError(f"State '{name}' is not defined", state_name=name)
    return self._states[name]

get_initial_state

get_initial_state() -> State

Get the initial leaf state.

Source code in src/pystator/machine.py

def get_initial_state(self) -> State:
    """Get the initial leaf state."""
    return self._states[self._engine.initial_leaf]

get_available_transitions

get_available_transitions(
    current_state: str,
) -> list[Transition]

Get all transitions available from a state.

Source code in src/pystator/machine.py

def get_available_transitions(self, current_state: str) -> list[Transition]:
    """Get all transitions available from a state."""
    ancestors_set = set(self._engine.hierarchy.ancestors(current_state))
    return [t for t in self._transitions if t.source & ancestors_set]

get_available_triggers

get_available_triggers(current_state: str) -> list[str]

Get all trigger names available from a state.

Source code in src/pystator/machine.py

def get_available_triggers(self, current_state: str) -> list[str]:
    """Get all trigger names available from a state."""
    transitions = self.get_available_transitions(current_state)
    return sorted({t.trigger for t in transitions})

can_transition

can_transition(
    current_state: str,
    trigger: str,
    context: dict[str, Any] | None = None,
) -> bool

Check if a transition is possible (convenience method).

Source code in src/pystator/machine.py

def can_transition(
    self,
    current_state: str,
    trigger: str,
    context: dict[str, Any] | None = None,
) -> bool:
    """Check if a transition is possible (convenience method)."""
    try:
        result = self.process(current_state, trigger, context)
        return result.success
    except Exception:
        return False

validate_state

validate_state(state: str) -> bool

True if state is defined.

Source code in src/pystator/machine.py

def validate_state(self, state: str) -> bool:
    """True if state is defined."""
    return state in self._states

is_terminal

is_terminal(state: str) -> bool

True if state is terminal.

Source code in src/pystator/machine.py

def is_terminal(self, state: str) -> bool:
    """True if state is terminal."""
    return self.get_state(state).is_terminal

is_initial

is_initial(state: str) -> bool

True if state is initial.

Source code in src/pystator/machine.py

def is_initial(self, state: str) -> bool:
    """True if state is initial."""
    return self.get_state(state).is_initial

get_state_variables

get_state_variables() -> list[dict[str, Any]]

Get state variable definitions from meta.

Source code in src/pystator/machine.py

def get_state_variables(self) -> list[dict[str, Any]]:
    """Get state variable definitions from meta."""
    raw = self._meta.get("state_variables")
    if not raw or not isinstance(raw, list):
        return []
    return [item if isinstance(item, dict) else {"key": str(item)} for item in raw]

validate_context

validate_context(
    context: dict[str, Any],
) -> tuple[bool, list[str]]

Validate context against state_variables.

Returns (valid, error_messages).

Source code in src/pystator/machine.py

def validate_context(self, context: dict[str, Any]) -> tuple[bool, list[str]]:
    """Validate context against state_variables.

    Returns (valid, error_messages).
    """
    if not self._meta.get("validate_context"):
        return True, []
    variables = self.get_state_variables()
    errors: list[str] = []
    for var in variables:
        key = var.get("key", "")
        if not key:
            continue
        required = var.get("required", False)
        type_hint = var.get("type")
        if key not in context:
            if required:
                errors.append(f"Missing required state variable: {key!r}")
            continue
        value = context[key]
        if required and value is None:
            errors.append(f"Required state variable {key!r} must not be None")
            continue
        if type_hint and value is not None:
            _TYPE_CHECKS = {
                "string": str,
                "number": (int, float),
                "boolean": bool,
                "object": dict,
            }
            expected = _TYPE_CHECKS.get(type_hint)
            if expected and not isinstance(value, expected):
                errors.append(
                    f"State variable {key!r} should be {type_hint}, "
                    f"got {type(value).__name__}"
                )
    return len(errors) == 0, errors

to_mermaid

to_mermaid(**kwargs: Any) -> str

Generate Mermaid stateDiagram-v2. See :func:visualization.to_mermaid.

Source code in src/pystator/machine.py

def to_mermaid(self, **kwargs: Any) -> str:
    """Generate Mermaid stateDiagram-v2. See :func:`visualization.to_mermaid`."""
    from pystator.visualization import to_mermaid

    return to_mermaid(self, **kwargs)

to_dot

to_dot(**kwargs: Any) -> str

Generate Graphviz DOT. See :func:visualization.to_dot.

Source code in src/pystator/machine.py

def to_dot(self, **kwargs: Any) -> str:
    """Generate Graphviz DOT. See :func:`visualization.to_dot`."""
    from pystator.visualization import to_dot

    return to_dot(self, **kwargs)

to_scxml

to_scxml(**kwargs: Any) -> str

Generate SCXML. See :func:visualization.to_scxml.

Source code in src/pystator/machine.py

def to_scxml(self, **kwargs: Any) -> str:
    """Generate SCXML. See :func:`visualization.to_scxml`."""
    from pystator.visualization import to_scxml

    return to_scxml(self, **kwargs)

get_statistics

get_statistics() -> dict[str, Any]

Get machine statistics. See :func:visualization.get_statistics.

Source code in src/pystator/machine.py

def get_statistics(self) -> dict[str, Any]:
    """Get machine statistics. See :func:`visualization.get_statistics`."""
    from pystator.visualization import get_statistics

    return get_statistics(self)

lint

lint() -> list[Any]

Run static analysis checks. See :func:lint.lint.

Source code in src/pystator/machine.py

def lint(self) -> list[Any]:
    """Run static analysis checks. See :func:`lint.lint`."""
    from pystator.lint import lint

    return lint(self)

StateMachineBuilder

StateMachineBuilder(machine_name: str)

Fluent builder for StateMachine from Python code.

Source code in src/pystator/machine.py

def __init__(self, machine_name: str) -> None:
    self._machine_name = machine_name
    self._states: list[dict[str, Any]] = []
    self._transitions: list[dict[str, Any]] = []
    self._state_variables: list[dict[str, Any]] = []

add_state

add_state(
    name: str, *, type: str = "stable", **kwargs: Any
) -> StateMachineBuilder

Add a state. type: initial, stable, terminal, error.

Source code in src/pystator/machine.py

def add_state(
    self,
    name: str,
    *,
    type: str = "stable",
    **kwargs: Any,
) -> StateMachineBuilder:
    """Add a state. type: initial, stable, terminal, error."""
    self._states.append({"name": name, "type": type, **kwargs})
    return self

add_transition

add_transition(
    trigger: str,
    source: str | list[str],
    dest: str,
    **kwargs: Any
) -> StateMachineBuilder

Add a transition.

Source code in src/pystator/machine.py

def add_transition(
    self,
    trigger: str,
    source: str | list[str],
    dest: str,
    **kwargs: Any,
) -> StateMachineBuilder:
    """Add a transition."""
    src = source if isinstance(source, list) else [source]
    self._transitions.append(
        {"trigger": trigger, "source": src, "dest": dest, **kwargs}
    )
    return self

add_state_variable

add_state_variable(
    key: str,
    *,
    type: str | None = None,
    required: bool = False,
    default: Any = None,
    **kwargs: Any
) -> StateMachineBuilder

Add a state variable (context key) definition.

Source code in src/pystator/machine.py

def add_state_variable(
    self,
    key: str,
    *,
    type: str | None = None,
    required: bool = False,
    default: Any = None,
    **kwargs: Any,
) -> StateMachineBuilder:
    """Add a state variable (context key) definition."""
    v: dict[str, Any] = {"key": key, **kwargs}
    if type is not None:
        v["type"] = type
    if required:
        v["required"] = True
    if default is not None:
        v["default"] = default
    self._state_variables.append(v)
    return self

to_dict

to_dict() -> dict[str, Any]

Export to config dict.

Source code in src/pystator/machine.py

def to_dict(self) -> dict[str, Any]:
    """Export to config dict."""
    d: dict[str, Any] = {
        "meta": {"machine_name": self._machine_name},
        "states": self._states,
        "transitions": self._transitions,
    }
    if self._state_variables:
        d["state_variables"] = self._state_variables
    return d

from_dict classmethod

from_dict(config: dict[str, Any]) -> StateMachineBuilder

Create builder from config dict (loads state_variables).

Source code in src/pystator/machine.py

@classmethod
def from_dict(cls, config: dict[str, Any]) -> StateMachineBuilder:
    """Create builder from config dict (loads state_variables)."""
    meta = config.get("meta", {})
    name = meta.get("machine_name", "unnamed")
    b = cls(name)
    b._states = list(config.get("states", []))
    b._transitions = list(config.get("transitions", []))
    sv = config.get("state_variables")
    b._state_variables = list(sv) if sv else []
    return b

build

build() -> StateMachine

Build the StateMachine.

Source code in src/pystator/machine.py

def build(self) -> StateMachine:
    """Build the StateMachine."""
    return StateMachine.from_dict(self.to_dict())