Multi

multi

Multi-action YAML parser and executor.

parse_yaml

parse_yaml(path: Path) -> list[dict[str, Any]]

Parse a YAML config file and validate its structure.

Supports a top-level 'vars' key for variable definitions. Variables are substituted in all action parameters using {{var}} syntax. Environment variables are accessible via {{env.KEY}}.

Parameters:

Name	Type	Description	Default
path	Path	Path to the YAML config file.	required

Returns:

Type	Description
list[dict[str, Any]]	A list of action dicts, each with a single key (action type)
list[dict[str, Any]]	and a dict of parameters.

Raises:

Type	Description
MultiConfigError	If the config structure is invalid.

Source code in wavexis/multi.py

def parse_yaml(path: Path) -> list[dict[str, Any]]:
    """Parse a YAML config file and validate its structure.

    Supports a top-level 'vars' key for variable definitions. Variables
    are substituted in all action parameters using {{var}} syntax.
    Environment variables are accessible via {{env.KEY}}.

    Args:
        path: Path to the YAML config file.

    Returns:
        A list of action dicts, each with a single key (action type)
        and a dict of parameters.

    Raises:
        MultiConfigError: If the config structure is invalid.
    """
    valid_path = validate_path(path)
    if not valid_path.exists():
        raise MultiConfigError("file", f"Config file not found: {path}")
    try:
        raw = yaml.safe_load(valid_path.read_text(encoding="utf-8"))
    except OSError as e:
        raise MultiConfigError("file", f"Config file not found or unreadable: {e}") from e
    if not isinstance(raw, dict):
        raise MultiConfigError("root", "Config must be a YAML mapping")
    variables = raw.get("vars", {})
    if variables and not isinstance(variables, dict):
        raise MultiConfigError("vars", "vars must be a mapping of key-value pairs")
    actions = raw.get("actions")
    if not isinstance(actions, list):
        raise MultiConfigError("actions", "Must be a list of action mappings")
    parsed: list[dict[str, Any]] = []
    for i, item in enumerate(actions):
        if not isinstance(item, dict) or len(item) != 1:
            raise MultiConfigError(
                f"actions[{i}]",
                "Each action must be a mapping with exactly one key",
            )
        action_type = next(iter(item))
        action_params = item[action_type]
        if not isinstance(action_params, dict):
            raise MultiConfigError(
                f"actions[{i}].{action_type}",
                "Action parameters must be a mapping",
            )
        action_params = _substitute_variables(action_params, variables)
        parsed.append({action_type: action_params})
    return parsed

execute_actions async

execute_actions(actions: list[dict[str, Any]], backend: Any, parallel: bool = False, cache: ActionCache | None = None) -> list[Any]

Execute each action, reusing the same backend.

Parameters:

Name	Type	Description	Default
actions	list[dict[str, Any]]	List of action dicts from parse_yaml.	required
backend	Any	An launched AbstractBackend instance.	required
parallel	bool	If True, execute all actions concurrently using separate tabs.	False
cache	ActionCache | None	Optional ActionCache. If provided, cacheable actions (screenshot, dom, scrape, eval, cookies, headers) will be served from cache on repeated calls with same URL+params.	None

Returns:

Type	Description
list[Any]	List of results from each action, in the same order as actions.

Source code in wavexis/multi.py

async def execute_actions(
    actions: list[dict[str, Any]],
    backend: Any,
    parallel: bool = False,
    cache: ActionCache | None = None,
) -> list[Any]:
    """Execute each action, reusing the same backend.

    Args:
        actions: List of action dicts from parse_yaml.
        backend: An launched AbstractBackend instance.
        parallel: If True, execute all actions concurrently using separate tabs.
        cache: Optional ActionCache. If provided, cacheable actions
            (screenshot, dom, scrape, eval, cookies, headers) will
            be served from cache on repeated calls with same URL+params.

    Returns:
        List of results from each action, in the same order as actions.
    """
    if parallel:

        async def _run_in_tab(action_dict: dict[str, Any]) -> Any:
            action_type = next(iter(action_dict))
            params = action_dict[action_type]
            url = params.get("url", "about:blank")
            tab = await backend.new_tab_handle(url)
            try:
                return await _dispatch(action_type, params, tab, cache=cache)
            finally:
                await tab.close()

        tasks = [_run_in_tab(ad) for ad in actions]
        gathered = await asyncio.gather(*tasks, return_exceptions=True)

        failures: list[str] = []
        for i, result in enumerate(gathered):
            if isinstance(result, BaseException) and not isinstance(result, Exception):
                raise result
            if isinstance(result, Exception):
                action_type = next(iter(actions[i]))
                failures.append(f"actions[{i}] ({action_type}): {result}")

        if failures:
            raise WavexisError("One or more actions failed:\n" + "\n".join(failures))

        return gathered

    results: list[Any] = []
    for i, action_dict in enumerate(actions):
        action_type = next(iter(action_dict))
        params = action_dict[action_type]
        try:
            result = await _dispatch(action_type, params, backend, cache=cache)
        except WavexisError:
            raise
        except Exception as e:
            # Wrap non-wavexis exceptions (e.g. CDP CommandError, TypeError)
            # so the CLI presents a friendly message instead of a raw traceback.
            selector = params.get("selector") if isinstance(params, dict) else None
            location = f"actions[{i}] ({action_type}"
            if selector:
                location += f" selector={selector!r}"
            location += ")"
            message = f"{location} failed: {e}"
            # Add a targeted hint for the common "type/fill on a non-focusable
            # element" case (e.g. typing into an <h1>). This is a frequent
            # mistake when generating multi-step configs from a single selector.
            msg_lower = str(e).lower()
            if action_type in ("type", "fill") and "not focusable" in msg_lower:
                message += (
                    "\nHint: 'type' and 'fill' require a focusable element "
                    "(<input>, <textarea>, <select> or a contenteditable element). "
                    "Use a different selector for the type action, or use 'click' "
                    "for non-input elements."
                )
            raise WavexisError(message) from e
        results.append(result)
    return results