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.
    """
    if not path.exists():
        raise MultiConfigError("file", f"Config file not found: {path}")
    raw = yaml.safe_load(path.read_text(encoding="utf-8"))
    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]
        return await asyncio.gather(*tasks)

    results: list[Any] = []
    for action_dict in actions:
        action_type = next(iter(action_dict))
        params = action_dict[action_type]
        result = await _dispatch(action_type, params, backend, cache=cache)
        results.append(result)
    return results