Actions

Register and execute actions (on_enter, on_exit, transition actions).

ActionRegistry

ActionRegistry()

Registry for action functions (sync and async).

Example

registry = ActionRegistry() registry.register("notify", lambda ctx: print("notified")) registry.execute("notify", {"user": "alice"})

Source code in src/pystator/actions.py

def __init__(self) -> None:
    self._actions: dict[str, AnyActionFunc] = {}
    self._metadata: dict[str, dict[str, Any]] = {}
    self._async_actions: set[str] = set()
    self._lock = threading.Lock()

register

register(
    name: str,
    func: AnyActionFunc,
    metadata: dict[str, Any] | None = None,
) -> None

Register an action function. Thread-safe.

Source code in src/pystator/actions.py

def register(
    self,
    name: str,
    func: AnyActionFunc,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Register an action function. Thread-safe."""
    if not name:
        raise ValueError("Action name cannot be empty")
    with self._lock:
        if name in self._actions:
            raise ValueError(f"Action '{name}' is already registered")
        self._actions[name] = func
        self._metadata[name] = metadata or {}
        if asyncio.iscoroutinefunction(func) or (
            hasattr(func, "__call__")
            and asyncio.iscoroutinefunction(getattr(func, "__call__", None))
        ):
            self._async_actions.add(name)

unregister

unregister(name: str) -> None

Unregister an action function. Thread-safe.

Source code in src/pystator/actions.py

def unregister(self, name: str) -> None:
    """Unregister an action function. Thread-safe."""
    with self._lock:
        if name not in self._actions:
            raise ActionNotFoundError(
                f"Action '{name}' not found", action_name=name
            )
        del self._actions[name]
        del self._metadata[name]
        self._async_actions.discard(name)

get_metadata

get_metadata(name: str) -> dict[str, Any]

Get metadata for a registered action.

Source code in src/pystator/actions.py

def get_metadata(self, name: str) -> dict[str, Any]:
    """Get metadata for a registered action."""
    return dict(self._metadata.get(name, {}))

decorator

decorator(
    name: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> Callable[[AnyActionFunc], AnyActionFunc]

Decorator to register an action function.

Example

@registry.decorator() ... def notify_user(ctx: dict) -> None: ... send_email(ctx["user_email"], "Order updated")

Source code in src/pystator/actions.py

def decorator(
    self,
    name: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> Callable[[AnyActionFunc], AnyActionFunc]:
    """Decorator to register an action function.

    Example:
        >>> @registry.decorator()
        ... def notify_user(ctx: dict) -> None:
        ...     send_email(ctx["user_email"], "Order updated")
    """

    def inner(func: AnyActionFunc) -> AnyActionFunc:
        self.register(name or func.__name__, func, metadata)
        return func

    return inner

ActionExecutor

ActionExecutor(
    registry: ActionRegistry,
    stop_on_error: bool = False,
    log_execution: bool = True,
    default_mode: ExecutionMode = ExecutionMode.SEQUENTIAL,
    strict: bool = False,
)

Executes actions from transition results.

Supports sequential, parallel, and phased execution modes.

Example

executor = ActionExecutor(action_registry) execution = executor.execute(transition_result, context)

Source code in src/pystator/actions.py

def __init__(
    self,
    registry: ActionRegistry,
    stop_on_error: bool = False,
    log_execution: bool = True,
    default_mode: ExecutionMode = ExecutionMode.SEQUENTIAL,
    strict: bool = False,
) -> None:
    self.registry = registry
    self.stop_on_error = stop_on_error
    self.log_execution = log_execution
    self.default_mode = default_mode
    self.strict = strict

execute

execute(
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult

Execute all actions from a transition result sequentially.

Source code in src/pystator/actions.py

def execute(
    self,
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult:
    """Execute all actions from a transition result sequentially."""
    result = ExecutionResult(
        transition_result=transition_result,
        started_at=datetime.now(timezone.utc),
    )
    if not transition_result.success:
        result.completed_at = datetime.now(timezone.utc)
        return result

    for spec in transition_result.all_action_specs:
        action_result = self._execute_single(
            spec.name, context, spec.params if spec.has_params else None
        )
        result.action_results.append(action_result)
        if not action_result.success and self.stop_on_error:
            break

    result.completed_at = datetime.now(timezone.utc)
    return result

execute_specific

execute_specific(
    action_names: list[str] | tuple[str, ...],
    context: dict[str, Any],
) -> list[ActionResult]

Execute specific actions by name sequentially.

Source code in src/pystator/actions.py

def execute_specific(
    self,
    action_names: list[str] | tuple[str, ...],
    context: dict[str, Any],
) -> list[ActionResult]:
    """Execute specific actions by name sequentially."""
    results: list[ActionResult] = []
    for name in action_names:
        results.append(self._execute_single(name, context, None))
    return results

validate_actions_exist

validate_actions_exist(
    transition_result: TransitionResult,
) -> list[str]

Return list of action names that are not registered.

Source code in src/pystator/actions.py

def validate_actions_exist(self, transition_result: TransitionResult) -> list[str]:
    """Return list of action names that are not registered."""
    missing: list[str] = []
    for spec in transition_result.all_action_specs:
        if spec.name.startswith("_effect."):
            continue  # Declarative effects don't need registry
        if not self.registry.has(spec.name):
            missing.append(spec.name)
    return missing

async_execute async

async_execute(
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult

Execute all actions asynchronously in sequential order.

Source code in src/pystator/actions.py

async def async_execute(
    self,
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult:
    """Execute all actions asynchronously in sequential order."""
    result = ExecutionResult(
        transition_result=transition_result,
        started_at=datetime.now(timezone.utc),
    )
    if not transition_result.success:
        result.completed_at = datetime.now(timezone.utc)
        return result

    for spec in transition_result.all_action_specs:
        ar = await self._async_execute_single(
            spec.name, context, spec.params if spec.has_params else None
        )
        result.action_results.append(ar)
        if not ar.success and self.stop_on_error:
            break

    result.completed_at = datetime.now(timezone.utc)
    return result

async_execute_parallel async

async_execute_parallel(
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult

Execute all actions concurrently using asyncio.gather.

Source code in src/pystator/actions.py

async def async_execute_parallel(
    self,
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult:
    """Execute all actions concurrently using asyncio.gather."""
    result = ExecutionResult(
        transition_result=transition_result,
        execution_mode=ExecutionMode.PARALLEL,
        started_at=datetime.now(timezone.utc),
    )
    if not transition_result.success:
        result.completed_at = datetime.now(timezone.utc)
        return result

    specs = transition_result.all_action_specs
    if specs:
        tasks = [
            self._async_execute_single(
                s.name, context, s.params if s.has_params else None
            )
            for s in specs
        ]
        result.action_results = list(await asyncio.gather(*tasks))

    result.completed_at = datetime.now(timezone.utc)
    return result

async_execute_phased async

async_execute_phased(
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult

Execute actions in phases: exit -> transition -> enter (each parallel).

Source code in src/pystator/actions.py

async def async_execute_phased(
    self,
    transition_result: TransitionResult,
    context: dict[str, Any],
) -> ExecutionResult:
    """Execute actions in phases: exit -> transition -> enter (each parallel)."""
    result = ExecutionResult(
        transition_result=transition_result,
        execution_mode=ExecutionMode.PHASED,
        started_at=datetime.now(timezone.utc),
    )
    if not transition_result.success:
        result.completed_at = datetime.now(timezone.utc)
        return result

    phases = [
        transition_result.on_exit_actions,
        transition_result.actions_to_execute,
        transition_result.on_enter_actions,
    ]
    all_results: list[ActionResult] = []
    for phase_specs in phases:
        if not phase_specs:
            continue
        tasks = [
            self._async_execute_single(
                s.name, context, s.params if s.has_params else None
            )
            for s in phase_specs
        ]
        phase_results = await asyncio.gather(*tasks)
        all_results.extend(phase_results)
        if self.stop_on_error and any(not r.success for r in phase_results):
            break

    result.action_results = all_results
    result.completed_at = datetime.now(timezone.utc)
    return result

async_execute_with_mode async

async_execute_with_mode(
    transition_result: TransitionResult,
    context: dict[str, Any],
    mode: ExecutionMode = ExecutionMode.SEQUENTIAL,
) -> ExecutionResult

Execute with explicit mode (sequential, parallel, or phased).

Source code in src/pystator/actions.py

async def async_execute_with_mode(
    self,
    transition_result: TransitionResult,
    context: dict[str, Any],
    mode: ExecutionMode = ExecutionMode.SEQUENTIAL,
) -> ExecutionResult:
    """Execute with explicit mode (sequential, parallel, or phased)."""
    if mode == ExecutionMode.PARALLEL:
        return await self.async_execute_parallel(transition_result, context)
    if mode == ExecutionMode.PHASED:
        return await self.async_execute_phased(transition_result, context)
    return await self.async_execute(transition_result, context)

async_execute_specific_parallel async

async_execute_specific_parallel(
    action_names: list[str] | tuple[str, ...],
    context: dict[str, Any],
) -> list[ActionResult]

Execute specific actions in parallel by name.

Source code in src/pystator/actions.py

async def async_execute_specific_parallel(
    self,
    action_names: list[str] | tuple[str, ...],
    context: dict[str, Any],
) -> list[ActionResult]:
    """Execute specific actions in parallel by name."""
    if not action_names:
        return []
    tasks = [
        self._async_execute_single(name, context, None) for name in action_names
    ]
    return list(await asyncio.gather(*tasks))

async_execute_action_spec async

async_execute_action_spec(
    action: ActionSpec, context: dict[str, Any]
) -> ActionResult

Async execute a single ActionSpec.

Source code in src/pystator/actions.py

async def async_execute_action_spec(
    self,
    action: ActionSpec,
    context: dict[str, Any],
) -> ActionResult:
    """Async execute a single ActionSpec."""
    return await self._async_execute_single(
        action.name, context, action.params if action.has_params else None
    )

async_execute_action_specs_parallel async

async_execute_action_specs_parallel(
    actions: tuple[ActionSpec, ...] | list[ActionSpec],
    context: dict[str, Any],
) -> list[ActionResult]

Async execute multiple ActionSpecs in parallel.

Source code in src/pystator/actions.py

async def async_execute_action_specs_parallel(
    self,
    actions: tuple[ActionSpec, ...] | list[ActionSpec],
    context: dict[str, Any],
) -> list[ActionResult]:
    """Async execute multiple ActionSpecs in parallel."""
    if not actions:
        return []
    tasks = [self.async_execute_action_spec(a, context) for a in actions]
    return list(await asyncio.gather(*tasks))