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._accepts_kwargs: 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.

Re-registration replaces the existing action (allows overriding builtins). If the function accepts keyword arguments beyond ctx, parameters from the action spec are passed as **kwargs instead of injected into the context dict.

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.

    Re-registration replaces the existing action (allows overriding builtins).
    If the function accepts keyword arguments beyond ``ctx``, parameters
    from the action spec are passed as ``**kwargs`` instead of injected
    into the context dict.
    """
    if not name:
        raise ValueError("Action name cannot be empty")
    with self._lock:
        self._actions[name] = func
        self._metadata[name] = metadata or {}
        if asyncio.iscoroutinefunction(func) or (
            callable(func)
            and asyncio.iscoroutinefunction(getattr(func, "__call__", None))  # noqa: B004
        ):
            self._async_actions.add(name)
        else:
            self._async_actions.discard(name)
        # Introspect: does the function accept extra keyword args?
        try:
            sig = inspect.signature(func)
            has_extra = any(
                p.kind
                in (
                    inspect.Parameter.KEYWORD_ONLY,
                    inspect.Parameter.VAR_KEYWORD,
                )
                for p in sig.parameters.values()
            )
            if has_extra:
                self._accepts_kwargs.add(name)
            else:
                self._accepts_kwargs.discard(name)
        except (ValueError, TypeError):
            self._accepts_kwargs.discard(name)

accepts_kwargs

accepts_kwargs(name: str) -> bool

Return True if the named action accepts keyword arguments.

Source code in src/pystator/actions.py

def accepts_kwargs(self, name: str) -> bool:
    """Return True if the named action accepts keyword arguments."""
    return name in self._accepts_kwargs

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

get(name: str) -> AnyActionFunc

Look up a registered action callable by name.

Parameters:

Name	Type	Description	Default
name	str	The registered action name.	required

Returns:

Type	Description
AnyActionFunc	The action callable.

Raises:

Type	Description
ActionNotFoundError	If no action is registered under name.

Source code in src/pystator/actions.py

def get(self, name: str) -> AnyActionFunc:
    """Look up a registered action callable by name.

    Args:
        name: The registered action name.

    Returns:
        The action callable.

    Raises:
        ActionNotFoundError: If no action is registered under *name*.
    """
    if name not in self._actions:
        raise ActionNotFoundError(f"Action '{name}' not found", action_name=name)
    return self._actions[name]

has

has(name: str) -> bool

Check whether an action is registered.

Source code in src/pystator/actions.py

def has(self, name: str) -> bool:
    """Check whether an action is registered."""
    return name in self._actions

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, {}))

is_async

is_async(name: str) -> bool

Return True if the named action is an async function.

Source code in src/pystator/actions.py

def is_async(self, name: str) -> bool:
    """Return True if the named action is an async function."""
    return name in self._async_actions

execute

execute(name: str, context: dict[str, Any], raise_on_error: bool = False) -> ActionResult

Execute a single named action synchronously.

Parameters:

Name	Type	Description	Default
name	str	Registered action name.	required
context	dict[str, Any]	Context dict passed to the action callable.	required
raise_on_error	bool	If True, re-raise exceptions instead of wrapping.	False

Returns:

Type	Description
ActionResult	ActionResult indicating success or failure.

Source code in src/pystator/actions.py

def execute(
    self, name: str, context: dict[str, Any], raise_on_error: bool = False
) -> ActionResult:
    """Execute a single named action synchronously.

    Args:
        name: Registered action name.
        context: Context dict passed to the action callable.
        raise_on_error: If True, re-raise exceptions instead of wrapping.

    Returns:
        ActionResult indicating success or failure.
    """
    func = self.get(name)
    try:
        result = func(context)
        return ActionResult.ok(name, result)
    except Exception as e:
        if raise_on_error:
            raise
        return ActionResult.fail(name, e)

execute_all

execute_all(actions: tuple[str, ...] | list[str], context: dict[str, Any], stop_on_error: bool = False) -> list[ActionResult]

Execute multiple actions with stop-on-error control.

Parameters:

Name	Type	Description	Default
actions	tuple[str, ...] | list[str]	Sequence of action names to execute.	required
context	dict[str, Any]	Context dict passed to each action.	required
stop_on_error	bool	If True, halt after the first failure.	False

Returns:

Type	Description
list[ActionResult]	List of ActionResult in execution order.

Source code in src/pystator/actions.py

def execute_all(
    self,
    actions: tuple[str, ...] | list[str],
    context: dict[str, Any],
    stop_on_error: bool = False,
) -> list[ActionResult]:
    """Execute multiple actions with stop-on-error control.

    Args:
        actions: Sequence of action names to execute.
        context: Context dict passed to each action.
        stop_on_error: If True, halt after the first failure.

    Returns:
        List of ActionResult in execution order.
    """
    results: list[ActionResult] = []
    for action_name in actions:
        try:
            result = self.execute(action_name, context)
            results.append(result)
            if not result.success and stop_on_error:
                break
        except ActionNotFoundError as e:
            results.append(ActionResult.fail(action_name, e))
            if stop_on_error:
                break
    return results

async_execute async

async_execute(name: str, context: dict[str, Any], raise_on_error: bool = False) -> ActionResult

Execute a single named action asynchronously.

Falls back to synchronous execution if the action is not async.

Parameters:

Name	Type	Description	Default
name	str	Registered action name.	required
context	dict[str, Any]	Context dict passed to the action callable.	required
raise_on_error	bool	If True, re-raise exceptions instead of wrapping.	False

Returns:

Type	Description
ActionResult	ActionResult indicating success or failure.

Source code in src/pystator/actions.py

async def async_execute(
    self, name: str, context: dict[str, Any], raise_on_error: bool = False
) -> ActionResult:
    """Execute a single named action asynchronously.

    Falls back to synchronous execution if the action is not async.

    Args:
        name: Registered action name.
        context: Context dict passed to the action callable.
        raise_on_error: If True, re-raise exceptions instead of wrapping.

    Returns:
        ActionResult indicating success or failure.
    """
    func = self.get(name)
    try:
        if self.is_async(name):
            result = await func(context)  # type: ignore[misc]
        else:
            result = func(context)
        return ActionResult.ok(name, result)
    except Exception as e:
        if raise_on_error:
            raise
        return ActionResult.fail(name, e)

async_execute_all async

async_execute_all(actions: tuple[str, ...] | list[str], context: dict[str, Any], stop_on_error: bool = False) -> list[ActionResult]

Execute multiple actions asynchronously.

Parameters:

Name	Type	Description	Default
actions	tuple[str, ...] | list[str]	Sequence of action names to execute.	required
context	dict[str, Any]	Context dict passed to each action.	required
stop_on_error	bool	If True, halt after the first failure.	False

Returns:

Type	Description
list[ActionResult]	List of ActionResult in execution order.

Source code in src/pystator/actions.py

async def async_execute_all(
    self,
    actions: tuple[str, ...] | list[str],
    context: dict[str, Any],
    stop_on_error: bool = False,
) -> list[ActionResult]:
    """Execute multiple actions asynchronously.

    Args:
        actions: Sequence of action names to execute.
        context: Context dict passed to each action.
        stop_on_error: If True, halt after the first failure.

    Returns:
        List of ActionResult in execution order.
    """
    results: list[ActionResult] = []
    for action_name in actions:
        try:
            result = await self.async_execute(action_name, context)
            results.append(result)
            if not result.success and stop_on_error:
                break
        except ActionNotFoundError as e:
            results.append(ActionResult.fail(action_name, e))
            if stop_on_error:
                break
    return results

list_actions

list_actions() -> list[str]

Return all registered action names.

Source code in src/pystator/actions.py

def list_actions(self) -> list[str]:
    """Return all registered action names."""
    return list(self._actions.keys())

clear

clear() -> None

Remove all registered actions.

Source code in src/pystator/actions.py

def clear(self) -> None:
    """Remove all registered actions."""
    self._actions.clear()
    self._metadata.clear()
    self._async_actions.clear()

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 module-attribute

ActionExecutor = _ActionExecutor