Skip to content

Actions

Base Classes

BaseAction

Bases: ABC, Generic[P, R]

Abstract base class for all wavexis actions.

An action encapsulates a single operation (e.g. screenshot, eval, pdf) that is executed against a backend.

Source code in wavexis/actions/base.py
class BaseAction(ABC, Generic[P, R]):
    """Abstract base class for all wavexis actions.

    An action encapsulates a single operation (e.g. screenshot, eval, pdf)
    that is executed against a backend.
    """

    def __init__(self, params: P) -> None:
        """Initialize the action with parameters.

        Args:
            params: Action-specific parameters.
        """
        self.params = params

    @abstractmethod
    async def execute(self, backend: AbstractBackend) -> R:
        """Execute the action against the given backend."""

__init__

__init__(params: P) -> None

Initialize the action with parameters.

Parameters:

Name Type Description Default
params P

Action-specific parameters.

required
Source code in wavexis/actions/base.py
def __init__(self, params: P) -> None:
    """Initialize the action with parameters.

    Args:
        params: Action-specific parameters.
    """
    self.params = params

execute abstractmethod async

execute(backend: AbstractBackend) -> R

Execute the action against the given backend.

Source code in wavexis/actions/base.py
@abstractmethod
async def execute(self, backend: AbstractBackend) -> R:
    """Execute the action against the given backend."""

NavigateAction

Bases: BaseAction[NavigateParams, None]

Action for navigating to a URL.

Source code in wavexis/actions/navigate.py
class NavigateAction(BaseAction[NavigateParams, None]):
    """Action for navigating to a URL."""

    async def execute(self, backend: AbstractBackend) -> None:
        """Execute the navigate action.

        Args:
            backend: The browser backend to use.
        """
        params = self.params
        await backend.navigate(params.url, params.wait)

execute async

execute(backend: AbstractBackend) -> None

Execute the navigate action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required
Source code in wavexis/actions/navigate.py
async def execute(self, backend: AbstractBackend) -> None:
    """Execute the navigate action.

    Args:
        backend: The browser backend to use.
    """
    params = self.params
    await backend.navigate(params.url, params.wait)

Screenshot & Capture

ScreenshotAction

Bases: BaseAction[ScreenshotParams, bytes]

Action for taking a screenshot of a web page.

Navigates to the URL in params, optionally executes JS, and captures a screenshot.

Source code in wavexis/actions/screenshot.py
class ScreenshotAction(BaseAction[ScreenshotParams, bytes]):
    """Action for taking a screenshot of a web page.

    Navigates to the URL in params, optionally executes JS, and captures
    a screenshot.
    """

    async def execute(self, backend: AbstractBackend) -> bytes:
        """Execute the screenshot action.

        Args:
            backend: The browser backend to use.

        Returns:
            Screenshot image bytes.
        """
        params = self.params
        await backend.navigate(params.url, params.wait)

        if params.js:
            await backend.eval(params.js, await_promise=True)

        if params.selector:
            return await backend.screenshot_selector(
                params.selector, params.format, params.quality
            )

        return await backend.screenshot(params)

execute async

execute(backend: AbstractBackend) -> bytes

Execute the screenshot action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
bytes

Screenshot image bytes.

Source code in wavexis/actions/screenshot.py
async def execute(self, backend: AbstractBackend) -> bytes:
    """Execute the screenshot action.

    Args:
        backend: The browser backend to use.

    Returns:
        Screenshot image bytes.
    """
    params = self.params
    await backend.navigate(params.url, params.wait)

    if params.js:
        await backend.eval(params.js, await_promise=True)

    if params.selector:
        return await backend.screenshot_selector(
            params.selector, params.format, params.quality
        )

    return await backend.screenshot(params)

ScreencastAction

Bases: BaseAction[ScreencastParams, list[str]]

Action for capturing screencast frames and saving them to a directory.

Source code in wavexis/actions/screencast.py
class ScreencastAction(BaseAction[ScreencastParams, list[str]]):
    """Action for capturing screencast frames and saving them to a directory."""

    def __init__(self, params: ScreencastParams, output_dir: str = "screencast") -> None:
        """Initialize the screencast action.

        Args:
            params: Screencast parameters including URL, format, and duration.
            output_dir: Directory to save captured frames.
        """
        self.params = params
        self._output_dir = output_dir

    async def execute(self, backend: AbstractBackend) -> list[str]:
        """Execute the screencast capture on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            List of saved frame file paths.
        """
        await backend.navigate(self.params.url, self.params.wait)
        frames = await backend.screencast(self.params)

        await asyncio.to_thread(
            lambda: Path(self._output_dir).mkdir(parents=True, exist_ok=True)
        )
        saved: list[str] = []
        for i, frame in enumerate(frames):
            ext = "png" if self.params.format == "png" else "jpg"
            fname = f"frame_{i:05d}.{ext}"
            fpath = str(Path(self._output_dir) / fname)
            await asyncio.to_thread(Path(fpath).write_bytes, frame)
            saved.append(fpath)
        return saved

__init__

__init__(params: ScreencastParams, output_dir: str = 'screencast') -> None

Initialize the screencast action.

Parameters:

Name Type Description Default
params ScreencastParams

Screencast parameters including URL, format, and duration.

required
output_dir str

Directory to save captured frames.

'screencast'
Source code in wavexis/actions/screencast.py
def __init__(self, params: ScreencastParams, output_dir: str = "screencast") -> None:
    """Initialize the screencast action.

    Args:
        params: Screencast parameters including URL, format, and duration.
        output_dir: Directory to save captured frames.
    """
    self.params = params
    self._output_dir = output_dir

execute async

execute(backend: AbstractBackend) -> list[str]

Execute the screencast capture on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
list[str]

List of saved frame file paths.

Source code in wavexis/actions/screencast.py
async def execute(self, backend: AbstractBackend) -> list[str]:
    """Execute the screencast capture on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        List of saved frame file paths.
    """
    await backend.navigate(self.params.url, self.params.wait)
    frames = await backend.screencast(self.params)

    await asyncio.to_thread(
        lambda: Path(self._output_dir).mkdir(parents=True, exist_ok=True)
    )
    saved: list[str] = []
    for i, frame in enumerate(frames):
        ext = "png" if self.params.format == "png" else "jpg"
        fname = f"frame_{i:05d}.{ext}"
        fpath = str(Path(self._output_dir) / fname)
        await asyncio.to_thread(Path(fpath).write_bytes, frame)
        saved.append(fpath)
    return saved

PDF & Print

PDFAction

Bases: BaseAction[PDFParams, bytes]

Action for generating a PDF of a web page.

Navigates to the URL in params, optionally executes JS, and generates a PDF.

Source code in wavexis/actions/pdf.py
class PDFAction(BaseAction[PDFParams, bytes]):
    """Action for generating a PDF of a web page.

    Navigates to the URL in params, optionally executes JS, and generates
    a PDF.
    """

    async def execute(self, backend: AbstractBackend) -> bytes:
        """Execute the PDF action.

        Args:
            backend: The browser backend to use.

        Returns:
            PDF bytes.
        """
        params = self.params
        await backend.navigate(params.url, params.wait)

        if params.js:
            await backend.eval(params.js, await_promise=True)

        return await backend.pdf(params)

execute async

execute(backend: AbstractBackend) -> bytes

Execute the PDF action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
bytes

PDF bytes.

Source code in wavexis/actions/pdf.py
async def execute(self, backend: AbstractBackend) -> bytes:
    """Execute the PDF action.

    Args:
        backend: The browser backend to use.

    Returns:
        PDF bytes.
    """
    params = self.params
    await backend.navigate(params.url, params.wait)

    if params.js:
        await backend.eval(params.js, await_promise=True)

    return await backend.pdf(params)

Evaluation & Scraping

EvalAction

Bases: BaseAction[EvalParams, Any]

Action for evaluating a JavaScript expression on a web page.

Navigates to the URL in params, then evaluates the expression. Supports @file syntax to read expression from a file.

Source code in wavexis/actions/eval.py
class EvalAction(BaseAction[EvalParams, Any]):
    """Action for evaluating a JavaScript expression on a web page.

    Navigates to the URL in params, then evaluates the expression.
    Supports @file syntax to read expression from a file.
    """

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the eval action.

        Args:
            backend: The browser backend to use.

        Returns:
            The JavaScript evaluation result.
        """
        params = self.params
        await backend.navigate(params.url, params.wait)

        expression = params.expression
        if params.file:
            file_path = params.file
            expression = await asyncio.to_thread(
                lambda: Path(file_path).read_text(encoding="utf-8")
            )

        return await backend.eval(expression, await_promise=params.await_promise)

execute async

execute(backend: AbstractBackend) -> Any

Execute the eval action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

The JavaScript evaluation result.

Source code in wavexis/actions/eval.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the eval action.

    Args:
        backend: The browser backend to use.

    Returns:
        The JavaScript evaluation result.
    """
    params = self.params
    await backend.navigate(params.url, params.wait)

    expression = params.expression
    if params.file:
        file_path = params.file
        expression = await asyncio.to_thread(
            lambda: Path(file_path).read_text(encoding="utf-8")
        )

    return await backend.eval(expression, await_promise=params.await_promise)

ScrapeAction

Bases: BaseAction[ScrapeParams, list[dict[str, Any]]]

Action for scraping multiple URLs.

Iterates over URLs, navigates to each, evaluates an expression, and collects results. Supports reading expression from a file via @file syntax.

Source code in wavexis/actions/scrape.py
class ScrapeAction(BaseAction[ScrapeParams, list[dict[str, Any]]]):
    """Action for scraping multiple URLs.

    Iterates over URLs, navigates to each, evaluates an expression,
    and collects results. Supports reading expression from a file via @file syntax.
    """

    async def execute(
        self, backend: AbstractBackend
    ) -> list[dict[str, Any]]:
        """Execute the scrape action.

        Args:
            backend: The browser backend to use.

        Returns:
            List of result dicts with "url" and "result" keys.
        """
        params = self.params
        expression = params.expression

        if params.file:
            file_path = params.file
            expression = await asyncio.to_thread(
                lambda: Path(file_path).read_text(encoding="utf-8")
            )

        if not expression:
            expression = "document.title"

        wait = params.wait
        if params.selector:
            wait = WaitStrategy(strategy="selector", selector=params.selector)

        results: list[dict[str, Any]] = []
        for url in params.urls:
            await backend.navigate(url, wait)
            value = await backend.eval(expression, await_promise=True)
            results.append({"url": url, "result": value})

        return results

execute async

execute(backend: AbstractBackend) -> list[dict[str, Any]]

Execute the scrape action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
list[dict[str, Any]]

List of result dicts with "url" and "result" keys.

Source code in wavexis/actions/scrape.py
async def execute(
    self, backend: AbstractBackend
) -> list[dict[str, Any]]:
    """Execute the scrape action.

    Args:
        backend: The browser backend to use.

    Returns:
        List of result dicts with "url" and "result" keys.
    """
    params = self.params
    expression = params.expression

    if params.file:
        file_path = params.file
        expression = await asyncio.to_thread(
            lambda: Path(file_path).read_text(encoding="utf-8")
        )

    if not expression:
        expression = "document.title"

    wait = params.wait
    if params.selector:
        wait = WaitStrategy(strategy="selector", selector=params.selector)

    results: list[dict[str, Any]] = []
    for url in params.urls:
        await backend.navigate(url, wait)
        value = await backend.eval(expression, await_promise=True)
        results.append({"url": url, "result": value})

    return results

DOM

DOMAction

Bases: BaseAction[DOMParams, Any]

Action for DOM operations.

Supports get (outer/inner HTML), query (single/all), attr get/set/remove, remove, focus, and scroll.

Source code in wavexis/actions/dom.py
class DOMAction(BaseAction[DOMParams, Any]):
    """Action for DOM operations.

    Supports get (outer/inner HTML), query (single/all), attr get/set/remove,
    remove, focus, and scroll.
    """

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the DOM action.

        Args:
            backend: The browser backend to use.

        Returns:
            str for "get" and "attr" actions, dict/list for "query",
            None for set/remove/focus/scroll.
        """
        params = self.params
        if params.url:
            await backend.navigate(params.url, params.wait)

        if params.action == "get":
            return await backend.dom_get(params.selector, outer=params.outer)

        if params.action == "query":
            return await backend.dom_query(params.selector, all=params.all)

        if params.action == "attr":
            if params.value is not None and params.attribute:
                await backend.dom_set_attr(
                    params.selector, params.attribute, params.value
                )
                return None
            if params.attribute:
                return await backend.dom_get_attr(params.selector, params.attribute)
            return None

        if params.action == "remove_attr":
            if params.attribute:
                await backend.dom_remove_attr(params.selector, params.attribute)
            return None

        if params.action == "remove":
            await backend.dom_remove(params.selector)
            return None

        if params.action == "focus":
            await backend.dom_focus(params.selector)
            return None

        if params.action == "scroll":
            await backend.dom_scroll(
                selector=params.selector or None,
                x=int(params.value or 0),
                y=int(params.attribute or 0) if params.attribute else 0,
            )
            return None

        return None

execute async

execute(backend: AbstractBackend) -> Any

Execute the DOM action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

str for "get" and "attr" actions, dict/list for "query",

Any

None for set/remove/focus/scroll.

Source code in wavexis/actions/dom.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the DOM action.

    Args:
        backend: The browser backend to use.

    Returns:
        str for "get" and "attr" actions, dict/list for "query",
        None for set/remove/focus/scroll.
    """
    params = self.params
    if params.url:
        await backend.navigate(params.url, params.wait)

    if params.action == "get":
        return await backend.dom_get(params.selector, outer=params.outer)

    if params.action == "query":
        return await backend.dom_query(params.selector, all=params.all)

    if params.action == "attr":
        if params.value is not None and params.attribute:
            await backend.dom_set_attr(
                params.selector, params.attribute, params.value
            )
            return None
        if params.attribute:
            return await backend.dom_get_attr(params.selector, params.attribute)
        return None

    if params.action == "remove_attr":
        if params.attribute:
            await backend.dom_remove_attr(params.selector, params.attribute)
        return None

    if params.action == "remove":
        await backend.dom_remove(params.selector)
        return None

    if params.action == "focus":
        await backend.dom_focus(params.selector)
        return None

    if params.action == "scroll":
        await backend.dom_scroll(
            selector=params.selector or None,
            x=int(params.value or 0),
            y=int(params.attribute or 0) if params.attribute else 0,
        )
        return None

    return None

DOMSnapshotAction

Bases: BaseAction[DOMSnapshotParams, dict[str, Any]]

Action for capturing a DOM snapshot of a web page.

Source code in wavexis/actions/dom_snapshot.py
class DOMSnapshotAction(BaseAction[DOMSnapshotParams, dict[str, Any]]):
    """Action for capturing a DOM snapshot of a web page."""

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the DOM snapshot action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict containing the raw DOM snapshot.
        """
        await backend.navigate(self.params.url, self.params.wait)
        return await backend.dom_snapshot()

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the DOM snapshot action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict containing the raw DOM snapshot.

Source code in wavexis/actions/dom_snapshot.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the DOM snapshot action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict containing the raw DOM snapshot.
    """
    await backend.navigate(self.params.url, self.params.wait)
    return await backend.dom_snapshot()

Input

InputAction

Bases: BaseAction[InputParams, Any]

Action for performing input interactions on a web page.

Source code in wavexis/actions/input.py
class InputAction(BaseAction[InputParams, Any]):
    """Action for performing input interactions on a web page."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the input action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            None for most actions; the result of the interaction.
        """
        await backend.navigate(self.params.url, self.params.wait)
        action = self.params.action
        if action == "click":
            await backend.click(
                self.params.selector,
                button=self.params.button,
                click_count=self.params.click_count,
            )
        elif action == "type":
            await backend.type_text(
                self.params.selector,
                self.params.text or "",
                delay=self.params.delay,
            )
        elif action == "fill":
            await backend.fill(
                self.params.selector, self.params.value or ""
            )
        elif action == "select":
            await backend.select_option(
                self.params.selector, self.params.value or ""
            )
        elif action == "hover":
            await backend.hover(self.params.selector)
        elif action == "key":
            await backend.key_press(self.params.key or "Enter")
        elif action == "drag":
            await backend.drag(
                self.params.source or "", self.params.target or ""
            )
        elif action == "tap":
            await backend.tap(self.params.selector)
        elif action == "scroll":
            await backend.dom_scroll(
                selector=self.params.selector or None,
                x=self.params.scroll_x,
                y=self.params.scroll_y,
            )
        elif action == "upload":
            await backend.set_files(
                self.params.selector, self.params.files or []
            )
        else:
            raise ValueError(f"Unknown input action: {action}")
        return None

execute async

execute(backend: AbstractBackend) -> Any

Execute the input action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

None for most actions; the result of the interaction.

Source code in wavexis/actions/input.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the input action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        None for most actions; the result of the interaction.
    """
    await backend.navigate(self.params.url, self.params.wait)
    action = self.params.action
    if action == "click":
        await backend.click(
            self.params.selector,
            button=self.params.button,
            click_count=self.params.click_count,
        )
    elif action == "type":
        await backend.type_text(
            self.params.selector,
            self.params.text or "",
            delay=self.params.delay,
        )
    elif action == "fill":
        await backend.fill(
            self.params.selector, self.params.value or ""
        )
    elif action == "select":
        await backend.select_option(
            self.params.selector, self.params.value or ""
        )
    elif action == "hover":
        await backend.hover(self.params.selector)
    elif action == "key":
        await backend.key_press(self.params.key or "Enter")
    elif action == "drag":
        await backend.drag(
            self.params.source or "", self.params.target or ""
        )
    elif action == "tap":
        await backend.tap(self.params.selector)
    elif action == "scroll":
        await backend.dom_scroll(
            selector=self.params.selector or None,
            x=self.params.scroll_x,
            y=self.params.scroll_y,
        )
    elif action == "upload":
        await backend.set_files(
            self.params.selector, self.params.files or []
        )
    else:
        raise ValueError(f"Unknown input action: {action}")
    return None

Console & Debugging

ConsoleAction

Bases: BaseAction[ConsoleParams, dict[str, Any]]

Action for capturing console messages and browser logs.

Navigates to the URL, then captures console messages and/or log entries.

Source code in wavexis/actions/console.py
class ConsoleAction(BaseAction[ConsoleParams, dict[str, Any]]):
    """Action for capturing console messages and browser logs.

    Navigates to the URL, then captures console messages and/or log entries.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the console action.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with "console" and/or "logs" keys containing entry lists.
        """
        params = self.params
        if params.url:
            await backend.navigate(params.url, params.wait)

        result: dict[str, Any] = {}

        if params.capture in ("console", "both"):
            result["console"] = await backend.capture_console(level=params.level)

        if params.capture in ("logs", "both"):
            result["logs"] = await backend.capture_logs()

        return result

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the console action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with "console" and/or "logs" keys containing entry lists.

Source code in wavexis/actions/console.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the console action.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with "console" and/or "logs" keys containing entry lists.
    """
    params = self.params
    if params.url:
        await backend.navigate(params.url, params.wait)

    result: dict[str, Any] = {}

    if params.capture in ("console", "both"):
        result["console"] = await backend.capture_console(level=params.level)

    if params.capture in ("logs", "both"):
        result["logs"] = await backend.capture_logs()

    return result

DebugAction

Bases: BaseAction[DebugActionParams, Any]

Action for debugging operations.

Source code in wavexis/actions/debug.py
class DebugAction(BaseAction[DebugActionParams, Any]):
    """Action for debugging operations."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the debug action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Result of the debug action (breakpoint ID, listener list, or None).

        Raises:
            ValueError: If the action is not recognized or required params missing.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)
        return await self._run_action(backend)

    async def _run_action(self, backend: AbstractBackend) -> Any:
        """Execute the debug action against the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Result of the debug operation.

        Raises:
            ValueError: If required parameters are missing for the action.
        """
        action = self.params.action
        if action == "breakpoint":
            if self.params.script_url is None or self.params.line is None:
                raise ValueError(
                    "script_url and line are required for breakpoint action"
                )
            return await backend.debug_set_breakpoint(
                self.params.script_url, self.params.line, self.params.condition
            )
        if action == "function_breakpoint":
            if not self.params.function_name:
                raise ValueError(
                    "function_name is required for function_breakpoint action"
                )
            return await backend.debug_set_breakpoint_function(
                self.params.function_name
            )
        if action == "remove_breakpoint":
            if not self.params.breakpoint_id:
                raise ValueError(
                    "breakpoint_id is required for remove_breakpoint action"
                )
            await backend.debug_remove_breakpoint(self.params.breakpoint_id)
            return None
        if action == "step_over":
            await backend.debug_step_over()
            return None
        if action == "step_into":
            await backend.debug_step_into()
            return None
        if action == "step_out":
            await backend.debug_step_out()
            return None
        if action == "pause":
            await backend.debug_pause()
            return None
        if action == "resume":
            await backend.debug_resume()
            return None
        if action == "listeners":
            if not self.params.selector:
                raise ValueError("selector is required for listeners action")
            return await backend.debug_get_listeners(self.params.selector)
        raise ValueError(f"Unknown debug action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the debug action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

Result of the debug action (breakpoint ID, listener list, or None).

Raises:

Type Description
ValueError

If the action is not recognized or required params missing.

Source code in wavexis/actions/debug.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the debug action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Result of the debug action (breakpoint ID, listener list, or None).

    Raises:
        ValueError: If the action is not recognized or required params missing.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)
    return await self._run_action(backend)

Performance & Profiling

PerformanceAction

Bases: BaseAction[PerformanceParams, dict[str, Any]]

Action for collecting performance data from a web page.

Source code in wavexis/actions/performance.py
class PerformanceAction(BaseAction[PerformanceParams, dict[str, Any]]):
    """Action for collecting performance data from a web page."""

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the performance action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict containing the performance data.

        Raises:
            ValueError: If the action is not recognized.
        """
        await backend.navigate(self.params.url, self.params.wait)
        return await self._run_action(backend)

    async def _run_action(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the performance action against the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Performance data as a dictionary.

        Raises:
            ValueError: If the action is not recognized.
        """
        action = self.params.action
        if action == "metrics":
            return await backend.perf_metrics()
        if action == "trace":
            return await backend.perf_trace(self.params.duration_ms)
        if action == "profile":
            return await backend.perf_profile(self.params.duration_ms)
        if action == "heap":
            return await backend.perf_heap_snapshot()
        if action == "coverage":
            return await backend.perf_coverage()
        if action == "css-coverage":
            return await backend.perf_css_coverage()
        raise ValueError(f"Unknown performance action: {action}")

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the performance action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict containing the performance data.

Raises:

Type Description
ValueError

If the action is not recognized.

Source code in wavexis/actions/performance.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the performance action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict containing the performance data.

    Raises:
        ValueError: If the action is not recognized.
    """
    await backend.navigate(self.params.url, self.params.wait)
    return await self._run_action(backend)

CoreWebVitalsAction

Bases: BaseAction[CoreWebVitalsParams, dict[str, Any]]

Action for measuring Core Web Vitals (LCP, CLS, INP) with scoring.

Collects LCP, CLS, INP, FCP, TTFB, TBT, and load time via PerformanceObserver and the Performance API. Computes ratings (good/needs-improvement/poor) and an overall score.

Source code in wavexis/actions/core_web_vitals.py
class CoreWebVitalsAction(BaseAction[CoreWebVitalsParams, dict[str, Any]]):
    """Action for measuring Core Web Vitals (LCP, CLS, INP) with scoring.

    Collects LCP, CLS, INP, FCP, TTFB, TBT, and load time via
    PerformanceObserver and the Performance API. Computes ratings
    (good/needs-improvement/poor) and an overall score.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the Core Web Vitals measurement.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with metrics, ratings, score, and budget check results.
        """
        await backend.navigate(self.params.url, self.params.wait)
        return await self._collect_cwv(backend)

    async def _collect_cwv(self, backend: AbstractBackend) -> dict[str, Any]:
        """Collect Core Web Vitals from the page.

        Args:
            backend: The launched browser backend.

        Returns:
            Dict with raw metrics, ratings, score, and budgets.
        """

        cwv_js = f"""
            (() => {{
                return new Promise(resolve => {{
                    let lcp = 0, cls = 0, inp = 0, tbt = 0;
                    new PerformanceObserver(list => {{
                        for (const e of list.getEntries()) {{
                            lcp = e.startTime;
                        }}
                    }}).observe({{type: 'largest-contentful-paint', buffered: true}});
                    new PerformanceObserver(list => {{
                        for (const e of list.getEntries()) {{
                            if (!e.hadRecentInput) cls += e.value;
                        }}
                    }}).observe({{type: 'layout-shift', buffered: true}});
                    new PerformanceObserver(list => {{
                        for (const e of list.getEntries()) {{
                            inp = Math.max(inp, e.duration);
                        }}
                    }}).observe({{type: 'event', buffered: true}});
                    new PerformanceObserver(list => {{
                        for (const e of list.getEntries()) {{
                            tbt += e.duration;
                        }}
                    }}).observe({{type: 'longtask', buffered: true}});
                    setTimeout(() => resolve({{lcp, cls, inp, tbt}}), {self.params.observe_ms});
                }});
            }})()
        """

        cwv: dict[str, Any] = {}
        with contextlib.suppress(Exception):
            cwv_result = await backend.eval(cwv_js, await_promise=True)
            if isinstance(cwv_result, dict):
                cwv = cwv_result

        timing_js = """
            (() => {
                const nav = performance.getEntriesByType('navigation')[0] || {};
                const paint = performance.getEntriesByType('paint');
                const fcp = paint.find(p => p.name === 'first-contentful-paint');
                return {
                    ttfb: nav.responseStart || 0,
                    fcp: fcp ? fcp.startTime : 0,
                    load: nav.loadEventEnd || 0,
                    domSize: document.querySelectorAll('*').length,
                    transferSize: nav.transferSize || 0,
                };
            })()
        """
        timing: dict[str, Any] = {}
        with contextlib.suppress(Exception):
            timing_result = await backend.eval(timing_js, await_promise=False)
            if isinstance(timing_result, dict):
                timing = timing_result

        lcp = cwv.get("lcp", 0)
        cls = cwv.get("cls", 0)
        inp = cwv.get("inp", 0)
        tbt = cwv.get("tbt", 0)
        ttfb = timing.get("ttfb", 0)
        fcp = timing.get("fcp", 0)
        load = timing.get("load", 0)
        dom_size = timing.get("domSize", 0)

        metrics: dict[str, float] = {
            "lcp_ms": lcp,
            "cls": cls,
            "inp_ms": inp,
            "fcp_ms": fcp,
            "ttfb_ms": ttfb,
            "tbt_ms": tbt,
            "load_ms": load,
        }

        ratings: dict[str, str] = {}
        for key, value in metrics.items():
            if key in THRESHOLDS:
                thresholds = THRESHOLDS[key]
                ratings[key] = _rating(
                    value, thresholds["good"], thresholds["poor"]
                )

        score = self._compute_score(metrics, dom_size)

        result: dict[str, Any] = {
            "url": self.params.url,
            "metrics": metrics,
            "ratings": ratings,
            "score": score,
            "dom_size": dom_size,
            "transfer_size": timing.get("transferSize", 0),
        }

        if self.params.budgets:
            result["budgets"] = self._check_budgets(metrics, self.params.budgets)

        return result

    def _compute_score(
        self, metrics: dict[str, float], dom_size: int
    ) -> int:
        """Compute a 0-100 score from Core Web Vitals metrics.

        Args:
            metrics: Dict of metric name → value.
            dom_size: Number of DOM elements.

        Returns:
            Score from 0 to 100.
        """
        score = 100
        for key, value in metrics.items():
            if key not in THRESHOLDS:
                continue
            thresholds = THRESHOLDS[key]
            if value > thresholds["poor"]:
                score -= 15
            elif value > thresholds["good"]:
                score -= 8
        if dom_size > 3000:
            score -= 10
        elif dom_size > 1500:
            score -= 5
        return max(0, score)

    def _check_budgets(
        self, metrics: dict[str, float], budgets: dict[str, float]
    ) -> dict[str, Any]:
        """Check metrics against budget thresholds.

        Args:
            metrics: Dict of metric name → value.
            budgets: Dict of metric name → max acceptable value.

        Returns:
            Dict with pass/fail status per metric and overall.
        """
        results: dict[str, Any] = {}
        all_pass = True
        for key, threshold in budgets.items():
            value = metrics.get(key, 0)
            passed = value <= threshold
            if not passed:
                all_pass = False
            results[key] = {
                "value": value,
                "budget": threshold,
                "pass": passed,
            }
        results["all_pass"] = all_pass
        return results

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the Core Web Vitals measurement.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with metrics, ratings, score, and budget check results.

Source code in wavexis/actions/core_web_vitals.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the Core Web Vitals measurement.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with metrics, ratings, score, and budget check results.
    """
    await backend.navigate(self.params.url, self.params.wait)
    return await self._collect_cwv(backend)

LighthouseAction

Bases: BaseAction[LighthouseParams, dict[str, Any]]

Action for running a browser-based audit using CDP metrics.

Instead of requiring the Lighthouse npm package, this action collects performance metrics, accessibility tree, and SEO meta tags directly via CDP/BiDi and computes scores.

Source code in wavexis/actions/lighthouse.py
class LighthouseAction(BaseAction[LighthouseParams, dict[str, Any]]):
    """Action for running a browser-based audit using CDP metrics.

    Instead of requiring the Lighthouse npm package, this action collects
    performance metrics, accessibility tree, and SEO meta tags directly
    via CDP/BiDi and computes scores.
    """

    async def execute(
        self, backend: AbstractBackend
    ) -> dict[str, Any]:
        """Execute the audit on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with category scores and detailed metrics.
        """
        await backend.navigate(self.params.url, self.params.wait)

        cats = self.params.categories or [
            "performance",
            "accessibility",
            "seo",
            "best-practices",
        ]
        result: dict[str, Any] = {
            "url": self.params.url,
            "categories": {},
        }

        if "performance" in cats:
            result["categories"]["performance"] = (
                await self._audit_performance(backend)
            )

        if "accessibility" in cats:
            result["categories"]["accessibility"] = (
                await self._audit_accessibility(backend)
            )

        if "seo" in cats:
            result["categories"]["seo"] = (
                await self._audit_seo(backend)
            )

        if "best-practices" in cats:
            result["categories"]["best-practices"] = (
                await self._audit_best_practices(backend)
            )

        return result

    async def _audit_performance(
        self, backend: AbstractBackend
    ) -> dict[str, Any]:
        """Collect performance metrics and compute a score."""
        metrics = await backend.perf_metrics()
        js = """
            (() => {
                const nav = performance.getEntriesByType('navigation')[0] || {};
                const paint = performance.getEntriesByType('paint');
                const fcp = paint.find(p => p.name === 'first-contentful-paint');
                return {
                    domContentLoaded: nav.domContentLoadedEventEnd || 0,
                    loadComplete: nav.loadEventEnd || 0,
                    ttfb: nav.responseStart || 0,
                    fcp: fcp ? fcp.startTime : 0,
                    domSize: document.querySelectorAll('*').length,
                    transferSize: nav.transferSize || 0,
                    encodedBodySize: nav.encodedBodySize || 0,
                };
            })()
        """
        perf = await backend.eval(js, await_promise=False)
        if not isinstance(perf, dict):
            perf = {}

        ttfb = perf.get("ttfb", 0)
        fcp = perf.get("fcp", 0)
        load = perf.get("loadComplete", 0)
        dom_size = perf.get("domSize", 0)

        cwv_js = """
            (() => {
                return new Promise(resolve => {
                    let lcp = 0, cls = 0, inp = 0, tbt = 0;
                    new PerformanceObserver(list => {
                        for (const e of list.getEntries()) {
                            lcp = e.startTime;
                        }
                    }).observe({type: 'largest-contentful-paint', buffered: true});
                    new PerformanceObserver(list => {
                        for (const e of list.getEntries()) {
                            if (!e.hadRecentInput) cls += e.value;
                        }
                    }).observe({type: 'layout-shift', buffered: true});
                    new PerformanceObserver(list => {
                        for (const e of list.getEntries()) {
                            inp = Math.max(inp, e.duration);
                        }
                    }).observe({type: 'event', buffered: true});
                    new PerformanceObserver(list => {
                        for (const e of list.getEntries()) {
                            tbt += e.duration;
                        }
                    }).observe({type: 'longtask', buffered: true});
                    setTimeout(() => resolve({lcp, cls, inp, tbt}), 500);
                });
            })()
        """
        cwv: dict[str, Any] = {}
        with contextlib.suppress(Exception):
            cwv_result = await backend.eval(cwv_js, await_promise=True)
            if isinstance(cwv_result, dict):
                cwv = cwv_result

        lcp = cwv.get("lcp", 0)
        cls = cwv.get("cls", 0)
        inp = cwv.get("inp", 0)
        tbt = cwv.get("tbt", 0)

        score = 100
        if ttfb > 800:
            score -= 10
        if ttfb > 1800:
            score -= 15
        if fcp > 1800:
            score -= 15
        if fcp > 3000:
            score -= 15
        if lcp > 2500:
            score -= 10
        if lcp > 4000:
            score -= 10
        if cls > 0.1:
            score -= 10
        if cls > 0.25:
            score -= 10
        if inp > 200:
            score -= 5
        if inp > 500:
            score -= 10
        if load > 3000:
            score -= 10
        if load > 5000:
            score -= 15
        if dom_size > 1500:
            score -= 10
        if dom_size > 3000:
            score -= 10
        score = max(0, score)

        result: dict[str, Any] = {
            "score": score,
            "ttfb_ms": ttfb,
            "fcp_ms": fcp,
            "lcp_ms": lcp,
            "cls": cls,
            "inp_ms": inp,
            "tbt_ms": tbt,
            "load_ms": load,
            "dom_size": dom_size,
            "transfer_size": perf.get("transferSize", 0),
            "raw_metrics": metrics,
        }

        if self.params.budgets:
            result["budgets"] = self._check_budgets(result, self.params.budgets)

        return result

    async def _audit_accessibility(
        self, backend: AbstractBackend
    ) -> dict[str, Any]:
        """Check common accessibility issues."""
        js = """
            (() => {
                const issues = [];
                const imgs = document.querySelectorAll('img');
                imgs.forEach(img => {
                    if (!img.alt) issues.push({type: 'image-alt', selector: getSelector(img)});
                });
                const inputs = document.querySelectorAll('input, textarea, select');
                inputs.forEach(input => {
                    if (!input.getAttribute('aria-label') &&
                        !input.labels?.length &&
                        !input.getAttribute('title')) {
                        issues.push({type: 'input-label', selector: getSelector(input)});
                    }
                });
                const links = document.querySelectorAll('a');
                links.forEach(link => {
                    if (!link.textContent.trim() && !link.getAttribute('aria-label')) {
                        issues.push({type: 'link-text', selector: getSelector(link)});
                    }
                });
                const buttons = document.querySelectorAll('button');
                buttons.forEach(btn => {
                    if (!btn.textContent.trim() && !btn.getAttribute('aria-label')) {
                        issues.push({type: 'button-text', selector: getSelector(btn)});
                    }
                });
                const html = document.documentElement;
                if (!html.getAttribute('lang')) {
                    issues.push({type: 'html-lang'});
                }
                function getSelector(el) {
                    if (el.id) return '#' + el.id;
                    return el.tagName.toLowerCase();
                }
                return {
                    issues: issues,
                    issue_count: issues.length,
                    has_lang: !!html.getAttribute('lang'),
                    has_viewport: !!document.querySelector('meta[name="viewport"]'),
                };
            })()
        """
        a11y = await backend.eval(js, await_promise=False)
        if not isinstance(a11y, dict):
            a11y = {"issues": [], "issue_count": 0}

        issue_count = a11y.get("issue_count", 0)
        score = max(0, 100 - issue_count * 5)

        return {
            "score": score,
            "issues": a11y.get("issues", []),
            "issue_count": issue_count,
            "has_lang": a11y.get("has_lang", False),
            "has_viewport": a11y.get("has_viewport", False),
        }

    async def _audit_seo(self, backend: AbstractBackend) -> dict[str, Any]:
        """Check SEO meta tags and content."""
        js = """
            (() => {
                const meta = (name) => {
                    const el = document.querySelector(`meta[name="${name}"]`) ||
                               document.querySelector(`meta[property="${name}"]`);
                    return el ? el.getAttribute('content') : null;
                };
                const title = document.title;
                const h1s = document.querySelectorAll('h1');
                const canonical = document.querySelector('link[rel="canonical"]');
                return {
                    title: title,
                    title_length: title.length,
                    description: meta('description'),
                    description_length: (meta('description') || '').length,
                    og_title: meta('og:title'),
                    og_description: meta('og:description'),
                    og_image: meta('og:image'),
                    twitter_card: meta('twitter:card'),
                    canonical: canonical ? canonical.getAttribute('href') : null,
                    h1_count: h1s.length,
                    has_robots_meta: !!document.querySelector('meta[name="robots"]'),
                    has_sitemap_link: !!document.querySelector('link[rel="sitemap"]'),
                };
            })()
        """
        seo = await backend.eval(js, await_promise=False)
        if not isinstance(seo, dict):
            seo = {}

        score = 100
        if not seo.get("title"):
            score -= 20
        elif len(seo.get("title", "")) > 60:
            score -= 5
        if not seo.get("description"):
            score -= 15
        elif len(seo.get("description", "")) > 160:
            score -= 5
        if seo.get("h1_count", 0) == 0:
            score -= 15
        if seo.get("h1_count", 0) > 1:
            score -= 5
        if not seo.get("canonical"):
            score -= 10
        if not seo.get("og_title"):
            score -= 10
        if not seo.get("twitter_card"):
            score -= 5
        score = max(0, score)

        return {"score": score, **seo}

    async def _audit_best_practices(
        self, backend: AbstractBackend
    ) -> dict[str, Any]:
        """Check best practices (HTTPS, console errors, deprecated APIs)."""
        js = """
            (() => {
                const issues = [];
                if (location.protocol !== 'https:') {
                    issues.push({type: 'not-https'});
                }
                const scripts = document.querySelectorAll('script[src]');
                scripts.forEach(s => {
                    if (s.src.startsWith('http://')) {
                        issues.push({type: 'insecure-script', src: s.src});
                    }
                });
                if (!window.isSecureContext) {
                    issues.push({type: 'insecure-context'});
                }
                return {
                    issues: issues,
                    is_https: location.protocol === 'https:',
                    console_errors: window.__wavexisConsoleErrors || [],
                };
            })()
        """
        console_errors: list[dict[str, Any]] = []
        with contextlib.suppress(WavexisError):
            console_errors = await backend.capture_console(level="error")

        bp = await backend.eval(js, await_promise=False)
        if not isinstance(bp, dict):
            bp = {"issues": [], "is_https": False}

        issues = bp.get("issues", [])
        score = 100
        for _ in issues:
            score -= 10
        score -= min(20, len(console_errors) * 5)
        score = max(0, score)

        return {
            "score": score,
            "issues": issues,
            "is_https": bp.get("is_https", False),
            "console_errors": console_errors,
        }

    @staticmethod
    def _check_budgets(
        metrics: dict[str, Any], budgets: dict[str, float]
    ) -> dict[str, Any]:
        """Check metrics against performance budgets.

        Args:
            metrics: Collected performance metrics.
            budgets: Dict of metric_name → max acceptable value.

        Returns:
            Dict with pass/fail per budget item and overall status.
        """
        results: list[dict[str, Any]] = []
        all_pass = True

        for key, threshold in budgets.items():
            actual = metrics.get(key, 0)
            if not isinstance(actual, (int, float)):
                actual = 0
            passed = actual <= threshold
            if not passed:
                all_pass = False
            results.append({
                "metric": key,
                "actual": actual,
                "budget": threshold,
                "pass": passed,
            })

        return {
            "pass": all_pass,
            "results": results,
        }

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the audit on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with category scores and detailed metrics.

Source code in wavexis/actions/lighthouse.py
async def execute(
    self, backend: AbstractBackend
) -> dict[str, Any]:
    """Execute the audit on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with category scores and detailed metrics.
    """
    await backend.navigate(self.params.url, self.params.wait)

    cats = self.params.categories or [
        "performance",
        "accessibility",
        "seo",
        "best-practices",
    ]
    result: dict[str, Any] = {
        "url": self.params.url,
        "categories": {},
    }

    if "performance" in cats:
        result["categories"]["performance"] = (
            await self._audit_performance(backend)
        )

    if "accessibility" in cats:
        result["categories"]["accessibility"] = (
            await self._audit_accessibility(backend)
        )

    if "seo" in cats:
        result["categories"]["seo"] = (
            await self._audit_seo(backend)
        )

    if "best-practices" in cats:
        result["categories"]["best-practices"] = (
            await self._audit_best_practices(backend)
        )

    return result

Network

NetworkAction

Bases: BaseAction[NetworkParams, Any]

Action for network operations.

Supports cookies get/set/delete/clear, headers, and user-agent override.

Source code in wavexis/actions/network.py
class NetworkAction(BaseAction[NetworkParams, Any]):
    """Action for network operations.

    Supports cookies get/set/delete/clear, headers, and user-agent override.
    """

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the network action.

        Args:
            backend: The browser backend to use.

        Returns:
            List of cookies for "cookies_get", None for other actions.
        """
        params = self.params

        if params.action == "cookies_get":
            return await backend.get_cookies()

        if params.action == "cookies_set":
            if params.cookie:
                await backend.set_cookie(params.cookie)
            elif params.cookies:
                from wavexis.config import CookieParams

                for cookie_dict in params.cookies:
                    cookie = CookieParams(
                        name=str(cookie_dict.get("name", "")),
                        value=str(cookie_dict.get("value", "")),
                        domain=str(cookie_dict.get("domain", "")),
                        path=str(cookie_dict.get("path", "/")),
                        secure=bool(cookie_dict.get("secure", True)),
                        http_only=bool(cookie_dict.get("http_only", False)),
                        same_site=str(cookie_dict.get("same_site", "Lax")),
                    )
                    await backend.set_cookie(cookie)
            return None

        if params.action == "cookies_delete":
            if params.name and params.domain:
                await backend.delete_cookie(params.name, params.domain)
            elif params.name and not params.domain:
                raise ValueError("domain is required for cookies_delete when name is specified")
            return None

        if params.action == "cookies_clear":
            await backend.clear_cookies()
            return None

        if params.action == "headers":
            if params.headers:
                await backend.set_headers(params.headers)
            return None

        if params.action == "user_agent":
            if params.user_agent:
                await backend.set_user_agent(params.user_agent)
            return None

        return None

execute async

execute(backend: AbstractBackend) -> Any

Execute the network action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

List of cookies for "cookies_get", None for other actions.

Source code in wavexis/actions/network.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the network action.

    Args:
        backend: The browser backend to use.

    Returns:
        List of cookies for "cookies_get", None for other actions.
    """
    params = self.params

    if params.action == "cookies_get":
        return await backend.get_cookies()

    if params.action == "cookies_set":
        if params.cookie:
            await backend.set_cookie(params.cookie)
        elif params.cookies:
            from wavexis.config import CookieParams

            for cookie_dict in params.cookies:
                cookie = CookieParams(
                    name=str(cookie_dict.get("name", "")),
                    value=str(cookie_dict.get("value", "")),
                    domain=str(cookie_dict.get("domain", "")),
                    path=str(cookie_dict.get("path", "/")),
                    secure=bool(cookie_dict.get("secure", True)),
                    http_only=bool(cookie_dict.get("http_only", False)),
                    same_site=str(cookie_dict.get("same_site", "Lax")),
                )
                await backend.set_cookie(cookie)
        return None

    if params.action == "cookies_delete":
        if params.name and params.domain:
            await backend.delete_cookie(params.name, params.domain)
        elif params.name and not params.domain:
            raise ValueError("domain is required for cookies_delete when name is specified")
        return None

    if params.action == "cookies_clear":
        await backend.clear_cookies()
        return None

    if params.action == "headers":
        if params.headers:
            await backend.set_headers(params.headers)
        return None

    if params.action == "user_agent":
        if params.user_agent:
            await backend.set_user_agent(params.user_agent)
        return None

    return None

HARAction

Bases: BaseAction[HarParams, dict[str, Any]]

Action for capturing HAR data.

Navigates to a URL, captures network traffic, and returns a HAR 1.2 dict.

Source code in wavexis/actions/har.py
class HARAction(BaseAction[HarParams, dict[str, Any]]):
    """Action for capturing HAR data.

    Navigates to a URL, captures network traffic, and returns a HAR 1.2 dict.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the HAR capture action.

        Args:
            backend: The browser backend to use.

        Returns:
            HAR 1.2 compliant dict with log.entries.
        """
        return await backend.capture_har(self.params)

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the HAR capture action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

HAR 1.2 compliant dict with log.entries.

Source code in wavexis/actions/har.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the HAR capture action.

    Args:
        backend: The browser backend to use.

    Returns:
        HAR 1.2 compliant dict with log.entries.
    """
    return await backend.capture_har(self.params)

HARReplayAction

Bases: BaseAction[HARReplayParams, dict[str, Any]]

Action for replaying network requests from a HAR file.

Navigates to the URL, then replays each request from the HAR file using the browser's fetch API.

Source code in wavexis/actions/har_replay.py
class HARReplayAction(BaseAction[HARReplayParams, dict[str, Any]]):
    """Action for replaying network requests from a HAR file.

    Navigates to the URL, then replays each request from the HAR file
    using the browser's fetch API.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the HAR replay action.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with replayed count and any errors.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)
        await backend.replay_har(self.params.har_path, self.params.url_filter)
        return {"status": "ok", "har_path": self.params.har_path}

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the HAR replay action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with replayed count and any errors.

Source code in wavexis/actions/har_replay.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the HAR replay action.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with replayed count and any errors.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)
    await backend.replay_har(self.params.har_path, self.params.url_filter)
    return {"status": "ok", "har_path": self.params.har_path}

ModifyRequestAction

Bases: BaseAction[ModifyRequestParams, dict[str, Any]]

Action for intercepting and modifying requests in-flight.

Sets up request interception with the given pattern and modifications, then navigates to the URL.

Source code in wavexis/actions/modify_request.py
class ModifyRequestAction(BaseAction[ModifyRequestParams, dict[str, Any]]):
    """Action for intercepting and modifying requests in-flight.

    Sets up request interception with the given pattern and modifications,
    then navigates to the URL.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the request modification action.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with status indicating interception was set up.
        """
        await backend.modify_request(self.params.pattern, self.params.modifications)
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)
        return {"status": "ok", "pattern": self.params.pattern}

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the request modification action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with status indicating interception was set up.

Source code in wavexis/actions/modify_request.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the request modification action.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with status indicating interception was set up.
    """
    await backend.modify_request(self.params.pattern, self.params.modifications)
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)
    return {"status": "ok", "pattern": self.params.pattern}

WebSocketInterceptAction

Bases: BaseAction[WebSocketParams, dict[str, Any]]

Action for intercepting WebSocket frames on a page.

Source code in wavexis/actions/websocket.py
class WebSocketInterceptAction(BaseAction[WebSocketParams, dict[str, Any]]):
    """Action for intercepting WebSocket frames on a page."""

    async def execute(
        self, backend: AbstractBackend
    ) -> dict[str, Any]:
        """Execute WebSocket interception on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with sent/received frames and connection info.
        """
        await backend.navigate(self.params.url, self.params.wait)

        js = f"""
            (() => {{
                const frames = {{sent: [], received: [], errors: []}};
                const pattern = {json.dumps(self.params.url_pattern)};
                const regex = pattern ? new RegExp(pattern) : null;
                const mocks = {json.dumps(self.params.mock_responses)};

                const origWS = window.WebSocket;
                window.WebSocket = function(url, protocols) {{
                    if (regex && !regex.test(url)) {{
                        return new origWS(url, protocols);
                    }}
                    const ws = new origWS(url, protocols);
                    frames.url = url;

                    const origSend = ws.send.bind(ws);
                    ws.send = function(data) {{
                        frames.sent.push({{timestamp: Date.now(), data: String(data)}});
                        if (mocks[String(data)]) {{
                            setTimeout(() => {{
                                frames.received.push({{
                                    timestamp: Date.now(),
                                    data: mocks[String(data)],
                                    mocked: true,
                                }});
                                ws.dispatchEvent(new MessageEvent(
                                    'message', {{data: mocks[String(data)]}}
                                ));
                            }}, 10);
                        }}
                        return origSend(data);
                    }};

                    ws.addEventListener('message', (e) => {{
                        frames.received.push({{timestamp: Date.now(), data: String(e.data)}});
                    }});
                    ws.addEventListener('error', (e) => {{
                        frames.errors.push({{timestamp: Date.now(), error: String(e)}});
                    }});
                    return ws;
                }};
                window.WebSocket.prototype = origWS.prototype;
                window.WebSocket.CONNECTING = origWS.CONNECTING;
                window.WebSocket.OPEN = origWS.OPEN;
                window.WebSocket.CLOSING = origWS.CLOSING;
                window.WebSocket.CLOSED = origWS.CLOSED;

                return new Promise((resolve) => {{
                    setTimeout(() => resolve(frames), {self.params.duration_ms});
                }});
            }})()
        """
        result = await backend.eval(js, await_promise=True)
        if isinstance(result, dict):
            return result
        return {"sent": [], "received": [], "errors": []}

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute WebSocket interception on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with sent/received frames and connection info.

Source code in wavexis/actions/websocket.py
async def execute(
    self, backend: AbstractBackend
) -> dict[str, Any]:
    """Execute WebSocket interception on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with sent/received frames and connection info.
    """
    await backend.navigate(self.params.url, self.params.wait)

    js = f"""
        (() => {{
            const frames = {{sent: [], received: [], errors: []}};
            const pattern = {json.dumps(self.params.url_pattern)};
            const regex = pattern ? new RegExp(pattern) : null;
            const mocks = {json.dumps(self.params.mock_responses)};

            const origWS = window.WebSocket;
            window.WebSocket = function(url, protocols) {{
                if (regex && !regex.test(url)) {{
                    return new origWS(url, protocols);
                }}
                const ws = new origWS(url, protocols);
                frames.url = url;

                const origSend = ws.send.bind(ws);
                ws.send = function(data) {{
                    frames.sent.push({{timestamp: Date.now(), data: String(data)}});
                    if (mocks[String(data)]) {{
                        setTimeout(() => {{
                            frames.received.push({{
                                timestamp: Date.now(),
                                data: mocks[String(data)],
                                mocked: true,
                            }});
                            ws.dispatchEvent(new MessageEvent(
                                'message', {{data: mocks[String(data)]}}
                            ));
                        }}, 10);
                    }}
                    return origSend(data);
                }};

                ws.addEventListener('message', (e) => {{
                    frames.received.push({{timestamp: Date.now(), data: String(e.data)}});
                }});
                ws.addEventListener('error', (e) => {{
                    frames.errors.push({{timestamp: Date.now(), error: String(e)}});
                }});
                return ws;
            }};
            window.WebSocket.prototype = origWS.prototype;
            window.WebSocket.CONNECTING = origWS.CONNECTING;
            window.WebSocket.OPEN = origWS.OPEN;
            window.WebSocket.CLOSING = origWS.CLOSING;
            window.WebSocket.CLOSED = origWS.CLOSED;

            return new Promise((resolve) => {{
                setTimeout(() => resolve(frames), {self.params.duration_ms});
            }});
        }})()
    """
    result = await backend.eval(js, await_promise=True)
    if isinstance(result, dict):
        return result
    return {"sent": [], "received": [], "errors": []}

Storage

StorageAction

Bases: BaseAction[StorageParams, Any]

Action for storage operations (DOM, Cache, IndexedDB).

Source code in wavexis/actions/storage.py
class StorageAction(BaseAction[StorageParams, Any]):
    """Action for storage operations (DOM, Cache, IndexedDB)."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the storage action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the storage operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "get":
            if not self.params.key:
                raise ValueError("key is required for get action")
            return await backend.storage_get(
                self.params.key, self.params.storage_type
            )

        if action == "set":
            if not self.params.key or self.params.value is None:
                raise ValueError("key and value are required for set action")
            await backend.storage_set(
                self.params.key,
                self.params.value,
                self.params.storage_type,
            )
            return None

        if action == "clear":
            await backend.storage_clear(self.params.storage_type)
            return None

        if action == "list":
            return await backend.storage_list(self.params.storage_type)

        if action == "cache-list":
            return await backend.cache_storage_list()

        if action == "cache-entries":
            if not self.params.cache_name:
                raise ValueError("cache_name is required for cache-entries action")
            return await backend.cache_storage_entries(self.params.cache_name)

        if action == "cache-delete":
            if not self.params.cache_name:
                raise ValueError("cache_name is required for cache-delete action")
            await backend.cache_storage_delete(self.params.cache_name)
            return None

        if action == "indexeddb-list":
            return await backend.indexeddb_list()

        if action == "indexeddb-get":
            if not self.params.database or not self.params.store:
                raise ValueError(
                    "database and store are required for indexeddb-get action"
                )
            return await backend.indexeddb_get_data(
                self.params.database, self.params.store
            )

        if action == "indexeddb-clear":
            if not self.params.database or not self.params.store:
                raise ValueError(
                    "database and store are required for indexeddb-clear action"
                )
            await backend.indexeddb_clear(self.params.database, self.params.store)
            return None

        raise ValueError(f"Unknown storage action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the storage action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the storage operation.

Source code in wavexis/actions/storage.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the storage action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the storage operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "get":
        if not self.params.key:
            raise ValueError("key is required for get action")
        return await backend.storage_get(
            self.params.key, self.params.storage_type
        )

    if action == "set":
        if not self.params.key or self.params.value is None:
            raise ValueError("key and value are required for set action")
        await backend.storage_set(
            self.params.key,
            self.params.value,
            self.params.storage_type,
        )
        return None

    if action == "clear":
        await backend.storage_clear(self.params.storage_type)
        return None

    if action == "list":
        return await backend.storage_list(self.params.storage_type)

    if action == "cache-list":
        return await backend.cache_storage_list()

    if action == "cache-entries":
        if not self.params.cache_name:
            raise ValueError("cache_name is required for cache-entries action")
        return await backend.cache_storage_entries(self.params.cache_name)

    if action == "cache-delete":
        if not self.params.cache_name:
            raise ValueError("cache_name is required for cache-delete action")
        await backend.cache_storage_delete(self.params.cache_name)
        return None

    if action == "indexeddb-list":
        return await backend.indexeddb_list()

    if action == "indexeddb-get":
        if not self.params.database or not self.params.store:
            raise ValueError(
                "database and store are required for indexeddb-get action"
            )
        return await backend.indexeddb_get_data(
            self.params.database, self.params.store
        )

    if action == "indexeddb-clear":
        if not self.params.database or not self.params.store:
            raise ValueError(
                "database and store are required for indexeddb-clear action"
            )
        await backend.indexeddb_clear(self.params.database, self.params.store)
        return None

    raise ValueError(f"Unknown storage action: {action}")

Emulation

EmulationAction

Bases: BaseAction[EmulationParams, None]

Action for browser emulation operations.

Supports device emulation, viewport override, geolocation override, timezone override, and dark mode emulation.

Source code in wavexis/actions/emulation.py
class EmulationAction(BaseAction[EmulationParams, None]):
    """Action for browser emulation operations.

    Supports device emulation, viewport override, geolocation override,
    timezone override, and dark mode emulation.
    """

    async def execute(self, backend: AbstractBackend) -> None:
        """Execute the emulation action on the backend.

        Args:
            backend: The backend to execute the action on.
        """
        params = self.params
        if params.action == "device":
            if params.device is None:
                raise ValueError("device is required for 'device' action")
            await backend.emulate_device(params.device)
        elif params.action == "viewport":
            await backend.set_viewport(
                params.width,
                params.height,
                params.device_scale_factor,
            )
        elif params.action == "geolocation":
            await backend.set_geolocation(
                params.latitude,
                params.longitude,
                params.accuracy,
            )
        elif params.action == "timezone":
            await backend.set_timezone(params.timezone)
        elif params.action == "dark_mode":
            await backend.set_dark_mode(params.dark_mode)
        else:
            raise ValueError(f"Unknown emulation action: {params.action}")

execute async

execute(backend: AbstractBackend) -> None

Execute the emulation action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The backend to execute the action on.

required
Source code in wavexis/actions/emulation.py
async def execute(self, backend: AbstractBackend) -> None:
    """Execute the emulation action on the backend.

    Args:
        backend: The backend to execute the action on.
    """
    params = self.params
    if params.action == "device":
        if params.device is None:
            raise ValueError("device is required for 'device' action")
        await backend.emulate_device(params.device)
    elif params.action == "viewport":
        await backend.set_viewport(
            params.width,
            params.height,
            params.device_scale_factor,
        )
    elif params.action == "geolocation":
        await backend.set_geolocation(
            params.latitude,
            params.longitude,
            params.accuracy,
        )
    elif params.action == "timezone":
        await backend.set_timezone(params.timezone)
    elif params.action == "dark_mode":
        await backend.set_dark_mode(params.dark_mode)
    else:
        raise ValueError(f"Unknown emulation action: {params.action}")

Permissions & Security

PermissionsAction

Bases: BaseAction[str, None]

Action for managing browser permissions.

Source code in wavexis/actions/permissions.py
class PermissionsAction(BaseAction[str, None]):
    """Action for managing browser permissions."""

    def __init__(
        self,
        params: str,
        action: str = "grant",
        permission: str = "",
        url: str = "",
        wait: WaitStrategy | None = None,
    ) -> None:
        """Initialize the permissions action.

        Args:
            params: Permissions parameters.
            action: Permission action ("grant", "deny", "reset", or "query").
            permission: Permission name (e.g. "geolocation", "notifications").
            url: URL to navigate to before the action.
            wait: Wait strategy after navigation.
        """
        self.params = params
        self._action = action
        self._permission = permission
        self._url = url
        self._wait = wait or WaitStrategy(strategy="load")

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the permissions action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            None.
        """
        if self._url:
            await backend.navigate(self._url, self._wait)
        if self._action == "grant":
            await backend.grant_permission(self._permission)
        elif self._action == "reset":
            await backend.reset_permissions()
        else:
            raise ValueError(f"Unknown permissions action: {self._action}")
        return None

__init__

__init__(params: str, action: str = 'grant', permission: str = '', url: str = '', wait: WaitStrategy | None = None) -> None

Initialize the permissions action.

Parameters:

Name Type Description Default
params str

Permissions parameters.

required
action str

Permission action ("grant", "deny", "reset", or "query").

'grant'
permission str

Permission name (e.g. "geolocation", "notifications").

''
url str

URL to navigate to before the action.

''
wait WaitStrategy | None

Wait strategy after navigation.

None
Source code in wavexis/actions/permissions.py
def __init__(
    self,
    params: str,
    action: str = "grant",
    permission: str = "",
    url: str = "",
    wait: WaitStrategy | None = None,
) -> None:
    """Initialize the permissions action.

    Args:
        params: Permissions parameters.
        action: Permission action ("grant", "deny", "reset", or "query").
        permission: Permission name (e.g. "geolocation", "notifications").
        url: URL to navigate to before the action.
        wait: Wait strategy after navigation.
    """
    self.params = params
    self._action = action
    self._permission = permission
    self._url = url
    self._wait = wait or WaitStrategy(strategy="load")

execute async

execute(backend: AbstractBackend) -> Any

Execute the permissions action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

None.

Source code in wavexis/actions/permissions.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the permissions action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        None.
    """
    if self._url:
        await backend.navigate(self._url, self._wait)
    if self._action == "grant":
        await backend.grant_permission(self._permission)
    elif self._action == "reset":
        await backend.reset_permissions()
    else:
        raise ValueError(f"Unknown permissions action: {self._action}")
    return None

SecurityAction

Bases: BaseAction[str, Any]

Action for security-related operations.

Source code in wavexis/actions/security.py
class SecurityAction(BaseAction[str, Any]):
    """Action for security-related operations."""

    def __init__(
        self,
        params: str,
        action: str = "state",
        ignore: bool = True,
        url: str = "",
        wait: WaitStrategy | None = None,
    ) -> None:
        """Initialize the security action.

        Args:
            params: Security parameters.
            action: Security action ("state" or "ignore-cert-errors").
            ignore: Whether to ignore certificate errors.
            url: URL to navigate to before the action.
            wait: Wait strategy after navigation.
        """
        self.params = params
        self._action = action
        self._ignore = ignore
        self._url = url
        self._wait = wait or WaitStrategy(strategy="load")

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the security action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Security state dict for "state" action; None for "ignore_cert".
        """
        if self._url:
            await backend.navigate(self._url, self._wait)
        if self._action == "state":
            return await backend.get_security_state()
        elif self._action == "ignore_cert":
            await backend.ignore_cert_errors(self._ignore)
            return None
        else:
            raise ValueError(f"Unknown security action: {self._action}")

__init__

__init__(params: str, action: str = 'state', ignore: bool = True, url: str = '', wait: WaitStrategy | None = None) -> None

Initialize the security action.

Parameters:

Name Type Description Default
params str

Security parameters.

required
action str

Security action ("state" or "ignore-cert-errors").

'state'
ignore bool

Whether to ignore certificate errors.

True
url str

URL to navigate to before the action.

''
wait WaitStrategy | None

Wait strategy after navigation.

None
Source code in wavexis/actions/security.py
def __init__(
    self,
    params: str,
    action: str = "state",
    ignore: bool = True,
    url: str = "",
    wait: WaitStrategy | None = None,
) -> None:
    """Initialize the security action.

    Args:
        params: Security parameters.
        action: Security action ("state" or "ignore-cert-errors").
        ignore: Whether to ignore certificate errors.
        url: URL to navigate to before the action.
        wait: Wait strategy after navigation.
    """
    self.params = params
    self._action = action
    self._ignore = ignore
    self._url = url
    self._wait = wait or WaitStrategy(strategy="load")

execute async

execute(backend: AbstractBackend) -> Any

Execute the security action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

Security state dict for "state" action; None for "ignore_cert".

Source code in wavexis/actions/security.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the security action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Security state dict for "state" action; None for "ignore_cert".
    """
    if self._url:
        await backend.navigate(self._url, self._wait)
    if self._action == "state":
        return await backend.get_security_state()
    elif self._action == "ignore_cert":
        await backend.ignore_cert_errors(self._ignore)
        return None
    else:
        raise ValueError(f"Unknown security action: {self._action}")

Dialog

DialogAction

Bases: BaseAction[str, None]

Action for handling JavaScript dialogs (alert, confirm, prompt).

Source code in wavexis/actions/dialog.py
class DialogAction(BaseAction[str, None]):
    """Action for handling JavaScript dialogs (alert, confirm, prompt)."""

    def __init__(
        self,
        params: str,
        action: str = "accept",
        prompt_text: str | None = None,
        url: str = "",
        wait: WaitStrategy | None = None,
    ) -> None:
        """Initialize the dialog action.

        Args:
            params: Dialog parameters or selector.
            action: Dialog action ("accept" or "dismiss").
            prompt_text: Text to enter in prompt dialogs.
            url: URL to navigate to before the action.
            wait: Wait strategy after navigation.
        """
        self.params = params
        self._action = action
        self._prompt_text = prompt_text
        self._url = url
        self._wait = wait or WaitStrategy(strategy="load")

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the dialog action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            None.
        """
        if self._url:
            await backend.navigate(self._url, self._wait)
        if self._action == "accept":
            await backend.dialog_accept(self._prompt_text)
        elif self._action == "dismiss":
            await backend.dialog_dismiss()
        else:
            raise ValueError(f"Unknown dialog action: {self._action}")
        return None

__init__

__init__(params: str, action: str = 'accept', prompt_text: str | None = None, url: str = '', wait: WaitStrategy | None = None) -> None

Initialize the dialog action.

Parameters:

Name Type Description Default
params str

Dialog parameters or selector.

required
action str

Dialog action ("accept" or "dismiss").

'accept'
prompt_text str | None

Text to enter in prompt dialogs.

None
url str

URL to navigate to before the action.

''
wait WaitStrategy | None

Wait strategy after navigation.

None
Source code in wavexis/actions/dialog.py
def __init__(
    self,
    params: str,
    action: str = "accept",
    prompt_text: str | None = None,
    url: str = "",
    wait: WaitStrategy | None = None,
) -> None:
    """Initialize the dialog action.

    Args:
        params: Dialog parameters or selector.
        action: Dialog action ("accept" or "dismiss").
        prompt_text: Text to enter in prompt dialogs.
        url: URL to navigate to before the action.
        wait: Wait strategy after navigation.
    """
    self.params = params
    self._action = action
    self._prompt_text = prompt_text
    self._url = url
    self._wait = wait or WaitStrategy(strategy="load")

execute async

execute(backend: AbstractBackend) -> Any

Execute the dialog action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

None.

Source code in wavexis/actions/dialog.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the dialog action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        None.
    """
    if self._url:
        await backend.navigate(self._url, self._wait)
    if self._action == "accept":
        await backend.dialog_accept(self._prompt_text)
    elif self._action == "dismiss":
        await backend.dialog_dismiss()
    else:
        raise ValueError(f"Unknown dialog action: {self._action}")
    return None

Download

DownloadAction

Bases: BaseAction[str, bytes]

Action for intercepting file downloads.

Source code in wavexis/actions/download.py
class DownloadAction(BaseAction[str, bytes]):
    """Action for intercepting file downloads."""

    def __init__(
        self,
        params: str,
        url: str = "",
        wait: WaitStrategy | None = None,
    ) -> None:
        """Initialize the download action.

        Args:
            params: Download parameters or URL filter.
            url: URL to navigate to before intercepting downloads.
            wait: Wait strategy after navigation.
        """
        self.params = params
        self._url = url
        self._wait = wait or WaitStrategy(strategy="load")

    async def execute(self, backend: AbstractBackend) -> bytes:
        """Execute the download interception on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Downloaded file bytes.
        """
        if self._url:
            await backend.navigate(self._url, self._wait)
        return await backend.intercept_download(self.params)

__init__

__init__(params: str, url: str = '', wait: WaitStrategy | None = None) -> None

Initialize the download action.

Parameters:

Name Type Description Default
params str

Download parameters or URL filter.

required
url str

URL to navigate to before intercepting downloads.

''
wait WaitStrategy | None

Wait strategy after navigation.

None
Source code in wavexis/actions/download.py
def __init__(
    self,
    params: str,
    url: str = "",
    wait: WaitStrategy | None = None,
) -> None:
    """Initialize the download action.

    Args:
        params: Download parameters or URL filter.
        url: URL to navigate to before intercepting downloads.
        wait: Wait strategy after navigation.
    """
    self.params = params
    self._url = url
    self._wait = wait or WaitStrategy(strategy="load")

execute async

execute(backend: AbstractBackend) -> bytes

Execute the download interception on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
bytes

Downloaded file bytes.

Source code in wavexis/actions/download.py
async def execute(self, backend: AbstractBackend) -> bytes:
    """Execute the download interception on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Downloaded file bytes.
    """
    if self._url:
        await backend.navigate(self._url, self._wait)
    return await backend.intercept_download(self.params)

Tabs

TabsAction

Bases: BaseAction[TabsParams, Any]

Action for tab management operations.

Supports listing, creating, closing, and activating tabs.

Source code in wavexis/actions/tabs.py
class TabsAction(BaseAction[TabsParams, Any]):
    """Action for tab management operations.

    Supports listing, creating, closing, and activating tabs.
    """

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the tabs action.

        Args:
            backend: The browser backend to use.

        Returns:
            List of tabs for "list", target ID str for "new", None otherwise.
        """
        params = self.params
        if params.action == "list":
            return await backend.list_tabs()
        if params.action == "new":
            return await backend.new_tab(params.url)
        if params.action == "close":
            await backend.close_tab(params.tab_id)
            return None
        if params.action == "activate":
            await backend.activate_tab(params.tab_id)
            return None
        return None

execute async

execute(backend: AbstractBackend) -> Any

Execute the tabs action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

List of tabs for "list", target ID str for "new", None otherwise.

Source code in wavexis/actions/tabs.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the tabs action.

    Args:
        backend: The browser backend to use.

    Returns:
        List of tabs for "list", target ID str for "new", None otherwise.
    """
    params = self.params
    if params.action == "list":
        return await backend.list_tabs()
    if params.action == "new":
        return await backend.new_tab(params.url)
    if params.action == "close":
        await backend.close_tab(params.tab_id)
        return None
    if params.action == "activate":
        await backend.activate_tab(params.tab_id)
        return None
    return None

Accessibility

AccessibilityAction

Bases: BaseAction[Any, Any]

Action for accessibility tree operations.

Source code in wavexis/actions/accessibility.py
class AccessibilityAction(BaseAction[Any, Any]):
    """Action for accessibility tree operations."""

    def __init__(
        self,
        params: Any,
        action: str = "tree",
        node_id: str = "",
        url: str = "",
        wait: WaitStrategy | None = None,
    ) -> None:
        """Initialize the accessibility action.

        Args:
            params: Action parameters.
            action: Accessibility action type ("tree" or "node").
            node_id: Node ID for node-specific actions.
            url: URL to navigate to before the action.
            wait: Wait strategy after navigation.
        """
        self.params = params
        self._action = action
        self._node_id = node_id
        self._url = url
        self._wait = wait or WaitStrategy(strategy="load")

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the accessibility action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Accessibility tree dict, node dict, or list of ancestor dicts.
        """
        if self._url:
            await backend.navigate(self._url, self._wait)
        if self._action == "tree":
            return await backend.a11y_tree()
        elif self._action == "node":
            return await backend.a11y_node(self._node_id)
        elif self._action == "ancestors":
            return await backend.a11y_ancestors(self._node_id)
        else:
            raise ValueError(f"Unknown a11y action: {self._action}")

__init__

__init__(params: Any, action: str = 'tree', node_id: str = '', url: str = '', wait: WaitStrategy | None = None) -> None

Initialize the accessibility action.

Parameters:

Name Type Description Default
params Any

Action parameters.

required
action str

Accessibility action type ("tree" or "node").

'tree'
node_id str

Node ID for node-specific actions.

''
url str

URL to navigate to before the action.

''
wait WaitStrategy | None

Wait strategy after navigation.

None
Source code in wavexis/actions/accessibility.py
def __init__(
    self,
    params: Any,
    action: str = "tree",
    node_id: str = "",
    url: str = "",
    wait: WaitStrategy | None = None,
) -> None:
    """Initialize the accessibility action.

    Args:
        params: Action parameters.
        action: Accessibility action type ("tree" or "node").
        node_id: Node ID for node-specific actions.
        url: URL to navigate to before the action.
        wait: Wait strategy after navigation.
    """
    self.params = params
    self._action = action
    self._node_id = node_id
    self._url = url
    self._wait = wait or WaitStrategy(strategy="load")

execute async

execute(backend: AbstractBackend) -> Any

Execute the accessibility action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

Accessibility tree dict, node dict, or list of ancestor dicts.

Source code in wavexis/actions/accessibility.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the accessibility action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Accessibility tree dict, node dict, or list of ancestor dicts.
    """
    if self._url:
        await backend.navigate(self._url, self._wait)
    if self._action == "tree":
        return await backend.a11y_tree()
    elif self._action == "node":
        return await backend.a11y_node(self._node_id)
    elif self._action == "ancestors":
        return await backend.a11y_ancestors(self._node_id)
    else:
        raise ValueError(f"Unknown a11y action: {self._action}")

CSS

CSSAction

Bases: BaseAction[CSSActionParams, dict[str, Any] | list[dict[str, Any]]]

Action for CSS inspection operations.

Source code in wavexis/actions/css.py
class CSSAction(BaseAction[CSSActionParams, dict[str, Any] | list[dict[str, Any]]]):
    """Action for CSS inspection operations."""

    async def execute(
        self, backend: AbstractBackend
    ) -> dict[str, Any] | list[dict[str, Any]]:
        """Execute the CSS action on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict or list of dicts containing CSS data.

        Raises:
            ValueError: If the action is not recognized or required params missing.
        """
        await backend.navigate(self.params.url, self.params.wait)
        return await self._run_action(backend)

    async def _run_action(
        self, backend: AbstractBackend
    ) -> dict[str, Any] | list[dict[str, Any]]:
        """Execute the CSS action against the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            CSS data as a dict or list of dicts depending on the action.

        Raises:
            ValueError: If required parameters are missing for the action.
        """
        action = self.params.action
        if action == "styles":
            if not self.params.selector:
                raise ValueError("selector is required for styles action")
            return await backend.css_get_styles(self.params.selector)
        if action == "stylesheets":
            return await backend.css_get_stylesheets()
        if action == "rules":
            if not self.params.stylesheet_id:
                raise ValueError("stylesheet_id is required for rules action")
            return await backend.css_get_rules(self.params.stylesheet_id)
        if action == "computed":
            if not self.params.selector:
                raise ValueError("selector is required for computed action")
            return await backend.css_get_computed(self.params.selector)
        raise ValueError(f"Unknown CSS action: {action}")

execute async

execute(backend: AbstractBackend) -> dict[str, Any] | list[dict[str, Any]]

Execute the CSS action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]]

Dict or list of dicts containing CSS data.

Raises:

Type Description
ValueError

If the action is not recognized or required params missing.

Source code in wavexis/actions/css.py
async def execute(
    self, backend: AbstractBackend
) -> dict[str, Any] | list[dict[str, Any]]:
    """Execute the CSS action on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict or list of dicts containing CSS data.

    Raises:
        ValueError: If the action is not recognized or required params missing.
    """
    await backend.navigate(self.params.url, self.params.wait)
    return await self._run_action(backend)

Overlay

OverlayAction

Bases: BaseAction[OverlayParams, None]

Action for overlay highlight and clear operations.

Source code in wavexis/actions/overlay.py
class OverlayAction(BaseAction[OverlayParams, None]):
    """Action for overlay highlight and clear operations."""

    async def execute(self, backend: AbstractBackend) -> None:
        """Execute the overlay action on the backend.

        Args:
            backend: The browser backend to use.

        Raises:
            ValueError: If the action is not recognized or selector is missing.
        """
        await backend.navigate(self.params.url, self.params.wait)
        await self._run_action(backend)

    async def _run_action(self, backend: AbstractBackend) -> None:
        """Execute the overlay action against the backend.

        Args:
            backend: The browser backend to use.

        Raises:
            ValueError: If required parameters are missing or action is unknown.
        """
        action = self.params.action
        if action == "highlight":
            if not self.params.selector:
                raise ValueError("selector is required for highlight action")
            await backend.overlay_highlight(self.params.selector, self.params.color)
        elif action == "clear":
            await backend.overlay_clear()
        else:
            raise ValueError(f"Unknown overlay action: {action}")

execute async

execute(backend: AbstractBackend) -> None

Execute the overlay action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Raises:

Type Description
ValueError

If the action is not recognized or selector is missing.

Source code in wavexis/actions/overlay.py
async def execute(self, backend: AbstractBackend) -> None:
    """Execute the overlay action on the backend.

    Args:
        backend: The browser backend to use.

    Raises:
        ValueError: If the action is not recognized or selector is missing.
    """
    await backend.navigate(self.params.url, self.params.wait)
    await self._run_action(backend)

Media

MediaAction

Bases: BaseAction[MediaParams, Any]

Action for Media operations (experimental).

Source code in wavexis/actions/media.py
class MediaAction(BaseAction[MediaParams, Any]):
    """Action for Media operations (experimental)."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the Media action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the Media operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "list":
            return await backend.media_get_players()

        if action == "messages":
            if not self.params.player_id:
                raise ValueError("player_id is required for messages action")
            return await backend.media_get_messages(self.params.player_id)

        raise ValueError(f"Unknown Media action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the Media action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the Media operation.

Source code in wavexis/actions/media.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the Media action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the Media operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "list":
        return await backend.media_get_players()

    if action == "messages":
        if not self.params.player_id:
            raise ValueError("player_id is required for messages action")
        return await backend.media_get_messages(self.params.player_id)

    raise ValueError(f"Unknown Media action: {action}")

Cast

CastAction

Bases: BaseAction[CastParams, Any]

Action for Cast operations (experimental).

Source code in wavexis/actions/cast.py
class CastAction(BaseAction[CastParams, Any]):
    """Action for Cast operations (experimental)."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the Cast action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the Cast operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "list":
            return await backend.cast_list()

        if action == "start-tab":
            if not self.params.sink_name:
                raise ValueError("sink_name is required for start-tab action")
            await backend.cast_start_tab(self.params.sink_name)
            return None

        if action == "stop":
            await backend.cast_stop()
            return None

        raise ValueError(f"Unknown Cast action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the Cast action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the Cast operation.

Source code in wavexis/actions/cast.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the Cast action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the Cast operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "list":
        return await backend.cast_list()

    if action == "start-tab":
        if not self.params.sink_name:
            raise ValueError("sink_name is required for start-tab action")
        await backend.cast_start_tab(self.params.sink_name)
        return None

    if action == "stop":
        await backend.cast_stop()
        return None

    raise ValueError(f"Unknown Cast action: {action}")

Bluetooth

BluetoothAction

Bases: BaseAction[BluetoothParams, Any]

Action for Bluetooth emulation operations (experimental).

Source code in wavexis/actions/bluetooth.py
class BluetoothAction(BaseAction[BluetoothParams, Any]):
    """Action for Bluetooth emulation operations (experimental)."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the Bluetooth action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the Bluetooth operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "emulate":
            if not self.params.name:
                raise ValueError("name is required for emulate action")
            await backend.bluetooth_emulate(
                self.params.name, self.params.address
            )
            return None

        if action == "stop":
            await backend.bluetooth_stop()
            return None

        raise ValueError(f"Unknown Bluetooth action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the Bluetooth action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the Bluetooth operation.

Source code in wavexis/actions/bluetooth.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the Bluetooth action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the Bluetooth operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "emulate":
        if not self.params.name:
            raise ValueError("name is required for emulate action")
        await backend.bluetooth_emulate(
            self.params.name, self.params.address
        )
        return None

    if action == "stop":
        await backend.bluetooth_stop()
        return None

    raise ValueError(f"Unknown Bluetooth action: {action}")

WebAudio

WebAudioAction

Bases: BaseAction[WebAudioParams, Any]

Action for WebAudio operations (experimental).

Source code in wavexis/actions/webaudio.py
class WebAudioAction(BaseAction[WebAudioParams, Any]):
    """Action for WebAudio operations (experimental)."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the WebAudio action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the WebAudio operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "list":
            return await backend.webaudio_get_contexts()

        if action == "get":
            if not self.params.context_id:
                raise ValueError("context_id is required for get action")
            return await backend.webaudio_get_context(self.params.context_id)

        raise ValueError(f"Unknown WebAudio action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the WebAudio action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the WebAudio operation.

Source code in wavexis/actions/webaudio.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the WebAudio action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the WebAudio operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "list":
        return await backend.webaudio_get_contexts()

    if action == "get":
        if not self.params.context_id:
            raise ValueError("context_id is required for get action")
        return await backend.webaudio_get_context(self.params.context_id)

    raise ValueError(f"Unknown WebAudio action: {action}")

WebAuthn

WebAuthnAction

Bases: BaseAction[WebAuthnParams, Any]

Action for WebAuthn virtual authenticator operations (experimental).

Source code in wavexis/actions/webauthn.py
class WebAuthnAction(BaseAction[WebAuthnParams, Any]):
    """Action for WebAuthn virtual authenticator operations (experimental)."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the WebAuthn action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the WebAuthn operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "add-virtual-authenticator":
            return await backend.webauthn_add_virtual_authenticator(
                self.params.protocol, self.params.transport
            )

        if action == "remove-authenticator":
            if not self.params.authenticator_id:
                raise ValueError(
                    "authenticator_id is required for remove-authenticator"
                )
            await backend.webauthn_remove_authenticator(
                self.params.authenticator_id
            )
            return None

        if action == "add-credential":
            if not self.params.authenticator_id or not self.params.credential:
                raise ValueError(
                    "authenticator_id and credential are required "
                    "for add-credential"
                )
            await backend.webauthn_add_credential(
                self.params.authenticator_id, self.params.credential
            )
            return None

        if action == "get-credentials":
            if not self.params.authenticator_id:
                raise ValueError(
                    "authenticator_id is required for get-credentials"
                )
            return await backend.webauthn_get_credentials(
                self.params.authenticator_id
            )

        raise ValueError(f"Unknown WebAuthn action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the WebAuthn action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the WebAuthn operation.

Source code in wavexis/actions/webauthn.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the WebAuthn action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the WebAuthn operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "add-virtual-authenticator":
        return await backend.webauthn_add_virtual_authenticator(
            self.params.protocol, self.params.transport
        )

    if action == "remove-authenticator":
        if not self.params.authenticator_id:
            raise ValueError(
                "authenticator_id is required for remove-authenticator"
            )
        await backend.webauthn_remove_authenticator(
            self.params.authenticator_id
        )
        return None

    if action == "add-credential":
        if not self.params.authenticator_id or not self.params.credential:
            raise ValueError(
                "authenticator_id and credential are required "
                "for add-credential"
            )
        await backend.webauthn_add_credential(
            self.params.authenticator_id, self.params.credential
        )
        return None

    if action == "get-credentials":
        if not self.params.authenticator_id:
            raise ValueError(
                "authenticator_id is required for get-credentials"
            )
        return await backend.webauthn_get_credentials(
            self.params.authenticator_id
        )

    raise ValueError(f"Unknown WebAuthn action: {action}")

Animation

AnimationAction

Bases: BaseAction[AnimationParams, Any]

Action for animation operations.

Source code in wavexis/actions/animation.py
class AnimationAction(BaseAction[AnimationParams, Any]):
    """Action for animation operations."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the animation action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the animation operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "list":
            return await backend.animation_list()

        if action == "pause":
            if not self.params.animation_id:
                raise ValueError("animation_id is required for pause action")
            await backend.animation_pause(self.params.animation_id)
            return None

        if action == "play":
            if not self.params.animation_id:
                raise ValueError("animation_id is required for play action")
            await backend.animation_play(self.params.animation_id)
            return None

        if action == "seek":
            if not self.params.animation_id or self.params.time_ms is None:
                raise ValueError(
                    "animation_id and time_ms are required for seek action"
                )
            await backend.animation_seek(
                self.params.animation_id, self.params.time_ms
            )
            return None

        raise ValueError(f"Unknown animation action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the animation action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the animation operation.

Source code in wavexis/actions/animation.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the animation action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the animation operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "list":
        return await backend.animation_list()

    if action == "pause":
        if not self.params.animation_id:
            raise ValueError("animation_id is required for pause action")
        await backend.animation_pause(self.params.animation_id)
        return None

    if action == "play":
        if not self.params.animation_id:
            raise ValueError("animation_id is required for play action")
        await backend.animation_play(self.params.animation_id)
        return None

    if action == "seek":
        if not self.params.animation_id or self.params.time_ms is None:
            raise ValueError(
                "animation_id and time_ms are required for seek action"
            )
        await backend.animation_seek(
            self.params.animation_id, self.params.time_ms
        )
        return None

    raise ValueError(f"Unknown animation action: {action}")

Service Worker

ServiceWorkerAction

Bases: BaseAction[ServiceWorkerParams, Any]

Action for service worker operations.

Source code in wavexis/actions/service_worker.py
class ServiceWorkerAction(BaseAction[ServiceWorkerParams, Any]):
    """Action for service worker operations."""

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the service worker action on the backend.

        Args:
            backend: An AbstractBackend instance.

        Returns:
            Result of the service worker operation.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        action = self.params.action

        if action == "list":
            return await backend.sw_list()

        if action == "unregister":
            if not self.params.registration_id:
                raise ValueError("registration_id is required for unregister action")
            await backend.sw_unregister(self.params.registration_id)
            return None

        if action == "update":
            if not self.params.registration_id:
                raise ValueError("registration_id is required for update action")
            await backend.sw_update(self.params.registration_id)
            return None

        raise ValueError(f"Unknown service worker action: {action}")

execute async

execute(backend: AbstractBackend) -> Any

Execute the service worker action on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An AbstractBackend instance.

required

Returns:

Type Description
Any

Result of the service worker operation.

Source code in wavexis/actions/service_worker.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the service worker action on the backend.

    Args:
        backend: An AbstractBackend instance.

    Returns:
        Result of the service worker operation.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    action = self.params.action

    if action == "list":
        return await backend.sw_list()

    if action == "unregister":
        if not self.params.registration_id:
            raise ValueError("registration_id is required for unregister action")
        await backend.sw_unregister(self.params.registration_id)
        return None

    if action == "update":
        if not self.params.registration_id:
            raise ValueError("registration_id is required for update action")
        await backend.sw_update(self.params.registration_id)
        return None

    raise ValueError(f"Unknown service worker action: {action}")

Multi-Action

MultiAction

Bases: BaseAction[Path, list[Any]]

Action that parses a YAML config and executes multiple actions.

Reuses a single backend instance for all actions. Supports both sequential (default) and parallel execution modes.

Source code in wavexis/actions/multi.py
class MultiAction(BaseAction[Path, list[Any]]):
    """Action that parses a YAML config and executes multiple actions.

    Reuses a single backend instance for all actions. Supports both
    sequential (default) and parallel execution modes.
    """

    def __init__(self, params: Path, parallel: bool = False) -> None:
        """Initialize the multi-action.

        Args:
            params: Path to the YAML config file.
            parallel: If True, execute actions concurrently.
        """
        super().__init__(params)
        self._parallel = parallel

    async def execute(self, backend: AbstractBackend) -> list[Any]:
        """Parse the YAML config and execute all actions on the backend.

        Args:
            backend: An already-launched AbstractBackend instance.

        Returns:
            List of results from each action.
        """
        actions = parse_yaml(self.params)
        return await execute_actions(actions, backend, parallel=self._parallel)

__init__

__init__(params: Path, parallel: bool = False) -> None

Initialize the multi-action.

Parameters:

Name Type Description Default
params Path

Path to the YAML config file.

required
parallel bool

If True, execute actions concurrently.

False
Source code in wavexis/actions/multi.py
def __init__(self, params: Path, parallel: bool = False) -> None:
    """Initialize the multi-action.

    Args:
        params: Path to the YAML config file.
        parallel: If True, execute actions concurrently.
    """
    super().__init__(params)
    self._parallel = parallel

execute async

execute(backend: AbstractBackend) -> list[Any]

Parse the YAML config and execute all actions on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

An already-launched AbstractBackend instance.

required

Returns:

Type Description
list[Any]

List of results from each action.

Source code in wavexis/actions/multi.py
async def execute(self, backend: AbstractBackend) -> list[Any]:
    """Parse the YAML config and execute all actions on the backend.

    Args:
        backend: An already-launched AbstractBackend instance.

    Returns:
        List of results from each action.
    """
    actions = parse_yaml(self.params)
    return await execute_actions(actions, backend, parallel=self._parallel)

Accessibility Audit

AxeAuditAction

Bases: BaseAction[AxeAuditParams, dict[str, Any]]

Action for running axe-core accessibility audit.

Navigates to the URL, injects axe-core, and runs the audit. Returns violations, passes, incomplete, and inapplicable results.

Source code in wavexis/actions/axe_audit.py
class AxeAuditAction(BaseAction[AxeAuditParams, dict[str, Any]]):
    """Action for running axe-core accessibility audit.

    Navigates to the URL, injects axe-core, and runs the audit.
    Returns violations, passes, incomplete, and inapplicable results.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the axe-core audit action.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with violations, passes, incomplete, and inapplicable lists.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)
        return await backend.axe_audit()

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the axe-core audit action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with violations, passes, incomplete, and inapplicable lists.

Source code in wavexis/actions/axe_audit.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the axe-core audit action.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with violations, passes, incomplete, and inapplicable lists.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)
    return await backend.axe_audit()

Crawl

CrawlAction

Bases: BaseAction[CrawlParams, list[dict[str, Any]]]

Action for crawling a website and collecting page data.

Source code in wavexis/actions/crawl.py
class CrawlAction(BaseAction[CrawlParams, list[dict[str, Any]]]):
    """Action for crawling a website and collecting page data."""

    async def execute(
        self, backend: AbstractBackend
    ) -> list[dict[str, Any]]:
        """Execute the crawl on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            List of page dicts with url, title, links, and depth.
        """
        results: list[dict[str, Any]] = []
        visited: set[str] = set()
        queue: list[tuple[str, int]] = [(self.params.start_url, 0)]
        origin = urlparse(self.params.start_url).netloc
        pattern = re.compile(self.params.url_pattern) if self.params.url_pattern else None

        while queue and len(results) < self.params.max_pages:
            url, depth = queue.pop(0)
            normalized = url.rstrip("/")
            if normalized in visited or depth > self.params.max_depth:
                continue
            visited.add(normalized)

            try:
                await backend.navigate(url, self.params.wait)
            except WavexisError:
                continue

            title = ""
            links: list[str] = []
            with contextlib.suppress(WavexisError):
                title = await backend.eval(
                    "document.title", await_promise=False
                )
            with contextlib.suppress(WavexisError):
                raw_links = await backend.eval(
                    "Array.from(document.querySelectorAll('a[href]')).map(a => a.href)",
                    await_promise=False,
                )
                if isinstance(raw_links, list):
                    links = [str(link) for link in raw_links if isinstance(link, str)]

            page_data: dict[str, Any] = {
                "url": url,
                "title": title,
                "depth": depth,
                "links_found": len(links),
            }
            results.append(page_data)

            if depth < self.params.max_depth:
                for link in links:
                    absolute = urljoin(url, link)
                    parsed = urlparse(absolute)
                    if parsed.scheme not in ("http", "https"):
                        continue
                    if self.params.same_origin and parsed.netloc != origin:
                        continue
                    if pattern and not pattern.search(absolute):
                        continue
                    norm = absolute.rstrip("/")
                    if norm not in visited:
                        queue.append((absolute, depth + 1))

        return results

execute async

execute(backend: AbstractBackend) -> list[dict[str, Any]]

Execute the crawl on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
list[dict[str, Any]]

List of page dicts with url, title, links, and depth.

Source code in wavexis/actions/crawl.py
async def execute(
    self, backend: AbstractBackend
) -> list[dict[str, Any]]:
    """Execute the crawl on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        List of page dicts with url, title, links, and depth.
    """
    results: list[dict[str, Any]] = []
    visited: set[str] = set()
    queue: list[tuple[str, int]] = [(self.params.start_url, 0)]
    origin = urlparse(self.params.start_url).netloc
    pattern = re.compile(self.params.url_pattern) if self.params.url_pattern else None

    while queue and len(results) < self.params.max_pages:
        url, depth = queue.pop(0)
        normalized = url.rstrip("/")
        if normalized in visited or depth > self.params.max_depth:
            continue
        visited.add(normalized)

        try:
            await backend.navigate(url, self.params.wait)
        except WavexisError:
            continue

        title = ""
        links: list[str] = []
        with contextlib.suppress(WavexisError):
            title = await backend.eval(
                "document.title", await_promise=False
            )
        with contextlib.suppress(WavexisError):
            raw_links = await backend.eval(
                "Array.from(document.querySelectorAll('a[href]')).map(a => a.href)",
                await_promise=False,
            )
            if isinstance(raw_links, list):
                links = [str(link) for link in raw_links if isinstance(link, str)]

        page_data: dict[str, Any] = {
            "url": url,
            "title": title,
            "depth": depth,
            "links_found": len(links),
        }
        results.append(page_data)

        if depth < self.params.max_depth:
            for link in links:
                absolute = urljoin(url, link)
                parsed = urlparse(absolute)
                if parsed.scheme not in ("http", "https"):
                    continue
                if self.params.same_origin and parsed.netloc != origin:
                    continue
                if pattern and not pattern.search(absolute):
                    continue
                norm = absolute.rstrip("/")
                if norm not in visited:
                    queue.append((absolute, depth + 1))

    return results

Visual Diff

VisualDiffAction

Bases: BaseAction[VisualDiffParams, dict[str, Any]]

Action for comparing a live screenshot against a baseline.

Captures a screenshot of the current page (or element), loads the baseline, and computes pixel-level differences.

Source code in wavexis/actions/visual_diff.py
class VisualDiffAction(BaseAction[VisualDiffParams, dict[str, Any]]):
    """Action for comparing a live screenshot against a baseline.

    Captures a screenshot of the current page (or element), loads the baseline,
    and computes pixel-level differences.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the visual diff action.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with diff_count, diff_percentage, total_pixels, and diff_base64.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        if self.params.selector:
            current_bytes = await backend.screenshot_selector(
                self.params.selector, format="png"
            )
        else:
            current_bytes = await backend.screenshot(
                ScreenshotParams(url="", format="png")
            )

        baseline_bytes = await asyncio.to_thread(
            Path(self.params.baseline_path).read_bytes
        )

        return self._compare(baseline_bytes, current_bytes)

    def _compare(
        self, baseline: bytes, current: bytes
    ) -> dict[str, Any]:
        """Compare two PNG byte buffers and return diff metrics.

        Args:
            baseline: Baseline PNG bytes.
            current: Current PNG bytes.

        Returns:
            Dict with diff_count, diff_percentage, total_pixels, and diff_base64.
        """
        try:
            from PIL import Image, ImageChops
        except ImportError:
            return {
                "error": "Pillow is required for visual diff. Install: pip install Pillow",
            }

        baseline_img = Image.open(io.BytesIO(baseline)).convert("RGB")
        current_img = Image.open(io.BytesIO(current)).convert("RGB")

        if baseline_img.size != current_img.size:
            current_img = current_img.resize(baseline_img.size)

        diff = ImageChops.difference(baseline_img, current_img)

        diff_array = list(diff.getdata())  # noqa: SIM118
        total_pixels = len(diff_array)
        diff_count = sum(
            1
            for r, g, b in diff_array
            if max(r, g, b) > self.params.threshold
        )

        diff_percentage = (diff_count / total_pixels * 100) if total_pixels else 0.0

        diff_img = diff.convert("L")
        diff_img = diff_img.point(lambda v: 255 if v > self.params.threshold else 0)
        diff_buffer = io.BytesIO()
        diff_img.save(diff_buffer, format="PNG")
        diff_b64 = base64.b64encode(diff_buffer.getvalue()).decode("ascii")

        return {
            "diff_count": diff_count,
            "diff_percentage": round(diff_percentage, 2),
            "total_pixels": total_pixels,
            "threshold": self.params.threshold,
            "diff_base64": diff_b64,
        }

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the visual diff action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with diff_count, diff_percentage, total_pixels, and diff_base64.

Source code in wavexis/actions/visual_diff.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the visual diff action.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with diff_count, diff_percentage, total_pixels, and diff_base64.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    if self.params.selector:
        current_bytes = await backend.screenshot_selector(
            self.params.selector, format="png"
        )
    else:
        current_bytes = await backend.screenshot(
            ScreenshotParams(url="", format="png")
        )

    baseline_bytes = await asyncio.to_thread(
        Path(self.params.baseline_path).read_bytes
    )

    return self._compare(baseline_bytes, current_bytes)

Combined Trace

CombinedTraceAction

Bases: BaseAction[CombinedTraceParams, dict[str, Any]]

Action for starting or stopping a combined trace.

Start: navigates to URL, starts combined trace, waits duration_ms, stops, and returns all collected data in one call. Stop: stops a previously started trace by ID.

Source code in wavexis/actions/combined_trace.py
class CombinedTraceAction(BaseAction[CombinedTraceParams, dict[str, Any]]):
    """Action for starting or stopping a combined trace.

    Start: navigates to URL, starts combined trace, waits duration_ms, stops,
    and returns all collected data in one call.
    Stop: stops a previously started trace by ID.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the combined trace action.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with trace data (trace_events, screenshots, network, console).
        """
        import asyncio

        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)

        if self.params.action == "start":
            trace_id = await backend.start_combined_trace(
                capture_screenshots=self.params.capture_screenshots,
                capture_network=self.params.capture_network,
                capture_console=self.params.capture_console,
            )
            await asyncio.sleep(self.params.duration_ms / 1000)
            return await backend.stop_combined_trace(trace_id)

        if self.params.action == "stop" and self.params.trace_id:
            return await backend.stop_combined_trace(self.params.trace_id)

        return {"error": f"Unknown action: {self.params.action}"}

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the combined trace action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with trace data (trace_events, screenshots, network, console).

Source code in wavexis/actions/combined_trace.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the combined trace action.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with trace data (trace_events, screenshots, network, console).
    """
    import asyncio

    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)

    if self.params.action == "start":
        trace_id = await backend.start_combined_trace(
            capture_screenshots=self.params.capture_screenshots,
            capture_network=self.params.capture_network,
            capture_console=self.params.capture_console,
        )
        await asyncio.sleep(self.params.duration_ms / 1000)
        return await backend.stop_combined_trace(trace_id)

    if self.params.action == "stop" and self.params.trace_id:
        return await backend.stop_combined_trace(self.params.trace_id)

    return {"error": f"Unknown action: {self.params.action}"}

Accessibility Audit

AxeAuditAction

Bases: BaseAction[AxeAuditParams, dict[str, Any]]

Action for running axe-core accessibility audit.

Navigates to the URL, injects axe-core, and runs the audit. Returns violations, passes, incomplete, and inapplicable results.

Source code in wavexis/actions/axe_audit.py
class AxeAuditAction(BaseAction[AxeAuditParams, dict[str, Any]]):
    """Action for running axe-core accessibility audit.

    Navigates to the URL, injects axe-core, and runs the audit.
    Returns violations, passes, incomplete, and inapplicable results.
    """

    async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
        """Execute the axe-core audit action.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with violations, passes, incomplete, and inapplicable lists.
        """
        if self.params.url:
            await backend.navigate(self.params.url, self.params.wait)
        return await backend.axe_audit()

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the axe-core audit action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with violations, passes, incomplete, and inapplicable lists.

Source code in wavexis/actions/axe_audit.py
async def execute(self, backend: AbstractBackend) -> dict[str, Any]:
    """Execute the axe-core audit action.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with violations, passes, incomplete, and inapplicable lists.
    """
    if self.params.url:
        await backend.navigate(self.params.url, self.params.wait)
    return await backend.axe_audit()

Browser

BrowserAction

Bases: BaseAction[str, Any]

Action for browser management operations.

Supports contexts (new/list/close), window bounds (get/set), and version. The params string specifies the action: "new_context", "list_contexts", "close_context", "get_window", "set_window", or "version".

Source code in wavexis/actions/browser.py
class BrowserAction(BaseAction[str, Any]):
    """Action for browser management operations.

    Supports contexts (new/list/close), window bounds (get/set), and version.
    The params string specifies the action: "new_context", "list_contexts",
    "close_context", "get_window", "set_window", or "version".
    """

    async def execute(self, backend: AbstractBackend) -> Any:
        """Execute the browser action.

        Args:
            backend: The browser backend to use.

        Returns:
            Action-dependent result (str, list, dict, or None).
        """
        action = self.params

        if action == "version":
            return await backend.browser_version()

        if action == "new_context":
            return await backend.new_context()

        if action == "list_contexts":
            return await backend.list_contexts()

        if action == "close_context":
            raise NotImplementedError("close_context action not yet implemented")

        if action == "get_window":
            raise NotImplementedError("get_window action not yet implemented")

        if action == "set_window":
            raise NotImplementedError("set_window action not yet implemented")

        return None

execute async

execute(backend: AbstractBackend) -> Any

Execute the browser action.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
Any

Action-dependent result (str, list, dict, or None).

Source code in wavexis/actions/browser.py
async def execute(self, backend: AbstractBackend) -> Any:
    """Execute the browser action.

    Args:
        backend: The browser backend to use.

    Returns:
        Action-dependent result (str, list, dict, or None).
    """
    action = self.params

    if action == "version":
        return await backend.browser_version()

    if action == "new_context":
        return await backend.new_context()

    if action == "list_contexts":
        return await backend.list_contexts()

    if action == "close_context":
        raise NotImplementedError("close_context action not yet implemented")

    if action == "get_window":
        raise NotImplementedError("get_window action not yet implemented")

    if action == "set_window":
        raise NotImplementedError("set_window action not yet implemented")

    return None

Session

SessionSaveAction

Bases: BaseAction[Path, str]

Action for saving browser session state to a file.

Source code in wavexis/actions/session.py
class SessionSaveAction(BaseAction[Path, str]):
    """Action for saving browser session state to a file."""

    async def execute(self, backend: AbstractBackend) -> str:
        """Save cookies and storage from the current backend session.

        Args:
            backend: The browser backend with an active session.

        Returns:
            JSON string of the session data.
        """
        cookies = await backend.get_cookies()
        local_storage = await backend.storage_list("local")
        session_storage = await backend.storage_list("session")
        url = ""
        with contextlib.suppress(WavexisError):
            url = await backend.eval("window.location.href", await_promise=False)

        data = SessionData(
            cookies=cookies,
            local_storage=dict(local_storage),
            session_storage=dict(session_storage),
            url=str(url) if url else "",
        )
        json_str = data.to_json()
        self.params.write_text(json_str, encoding="utf-8")
        return json_str

execute async

execute(backend: AbstractBackend) -> str

Save cookies and storage from the current backend session.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend with an active session.

required

Returns:

Type Description
str

JSON string of the session data.

Source code in wavexis/actions/session.py
async def execute(self, backend: AbstractBackend) -> str:
    """Save cookies and storage from the current backend session.

    Args:
        backend: The browser backend with an active session.

    Returns:
        JSON string of the session data.
    """
    cookies = await backend.get_cookies()
    local_storage = await backend.storage_list("local")
    session_storage = await backend.storage_list("session")
    url = ""
    with contextlib.suppress(WavexisError):
        url = await backend.eval("window.location.href", await_promise=False)

    data = SessionData(
        cookies=cookies,
        local_storage=dict(local_storage),
        session_storage=dict(session_storage),
        url=str(url) if url else "",
    )
    json_str = data.to_json()
    self.params.write_text(json_str, encoding="utf-8")
    return json_str

SessionLoadAction

Bases: BaseAction[Path, None]

Action for loading browser session state from a file.

Source code in wavexis/actions/session.py
class SessionLoadAction(BaseAction[Path, None]):
    """Action for loading browser session state from a file."""

    async def execute(self, backend: AbstractBackend) -> None:
        """Load cookies and storage into the backend session.

        Args:
            backend: The browser backend with an active session.
        """
        data = SessionData.from_json(
            self.params.read_text(encoding="utf-8")
        )

        for cookie in data.cookies:
            cp = CookieParams(
                name=cookie.get("name", ""),
                value=cookie.get("value", ""),
                domain=cookie.get("domain", ""),
                path=cookie.get("path", "/"),
                secure=cookie.get("secure", True),
                http_only=cookie.get("httpOnly", False),
                same_site=cookie.get("sameSite", "Lax"),
            )
            await backend.set_cookie(cp)

        for key, value in data.local_storage.items():
            await backend.storage_set(key, value, "local")
        for key, value in data.session_storage.items():
            await backend.storage_set(key, value, "session")

execute async

execute(backend: AbstractBackend) -> None

Load cookies and storage into the backend session.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend with an active session.

required
Source code in wavexis/actions/session.py
async def execute(self, backend: AbstractBackend) -> None:
    """Load cookies and storage into the backend session.

    Args:
        backend: The browser backend with an active session.
    """
    data = SessionData.from_json(
        self.params.read_text(encoding="utf-8")
    )

    for cookie in data.cookies:
        cp = CookieParams(
            name=cookie.get("name", ""),
            value=cookie.get("value", ""),
            domain=cookie.get("domain", ""),
            path=cookie.get("path", "/"),
            secure=cookie.get("secure", True),
            http_only=cookie.get("httpOnly", False),
            same_site=cookie.get("sameSite", "Lax"),
        )
        await backend.set_cookie(cp)

    for key, value in data.local_storage.items():
        await backend.storage_set(key, value, "local")
    for key, value in data.session_storage.items():
        await backend.storage_set(key, value, "session")

Extract

ExtractAction

Bases: BaseAction[ExtractParams, list[dict[str, Any]]]

Action for extracting structured data from a page using a CSS selector schema.

Source code in wavexis/actions/extract.py
class ExtractAction(BaseAction[ExtractParams, list[dict[str, Any]]]):
    """Action for extracting structured data from a page using a CSS selector schema."""

    async def execute(
        self, backend: AbstractBackend
    ) -> list[dict[str, Any]]:
        """Execute the extraction on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            List of dicts with field names mapped to extracted text/HTML.
        """
        await backend.navigate(self.params.url, self.params.wait)

        schema_json = json.dumps(self.params.schema)
        if self.params.selector:
            selector_json = json.dumps(self.params.selector)
            js = f"""
                (() => {{
                    const schema = {schema_json};
                    const elements = document.querySelectorAll({selector_json});
                    return Array.from(elements).map(el => {{
                        const row = {{}};
                        for (const [field, sel] of Object.entries(schema)) {{
                            const node = el.querySelector(sel);
                            row[field] = node ? node.textContent.trim() : null;
                        }}
                        return row;
                    }});
                }})()
            """
        else:
            js = f"""
                (() => {{
                    const schema = {schema_json};
                    const row = {{}};
                    for (const [field, sel] of Object.entries(schema)) {{
                        const node = document.querySelector(sel);
                        row[field] = node ? node.textContent.trim() : null;
                    }}
                    return [row];
                }})()
            """

        result = await backend.eval(js, await_promise=False)
        if isinstance(result, list):
            return result
        return []

execute async

execute(backend: AbstractBackend) -> list[dict[str, Any]]

Execute the extraction on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
list[dict[str, Any]]

List of dicts with field names mapped to extracted text/HTML.

Source code in wavexis/actions/extract.py
async def execute(
    self, backend: AbstractBackend
) -> list[dict[str, Any]]:
    """Execute the extraction on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        List of dicts with field names mapped to extracted text/HTML.
    """
    await backend.navigate(self.params.url, self.params.wait)

    schema_json = json.dumps(self.params.schema)
    if self.params.selector:
        selector_json = json.dumps(self.params.selector)
        js = f"""
            (() => {{
                const schema = {schema_json};
                const elements = document.querySelectorAll({selector_json});
                return Array.from(elements).map(el => {{
                    const row = {{}};
                    for (const [field, sel] of Object.entries(schema)) {{
                        const node = el.querySelector(sel);
                        row[field] = node ? node.textContent.trim() : null;
                    }}
                    return row;
                }});
            }})()
        """
    else:
        js = f"""
            (() => {{
                const schema = {schema_json};
                const row = {{}};
                for (const [field, sel] of Object.entries(schema)) {{
                    const node = document.querySelector(sel);
                    row[field] = node ? node.textContent.trim() : null;
                }}
                return [row];
            }})()
        """

    result = await backend.eval(js, await_promise=False)
    if isinstance(result, list):
        return result
    return []

Form

FormAction

Bases: BaseAction[FormParams, dict[str, Any]]

Action for filling form fields and optionally submitting.

Source code in wavexis/actions/form.py
class FormAction(BaseAction[FormParams, dict[str, Any]]):
    """Action for filling form fields and optionally submitting."""

    async def execute(
        self, backend: AbstractBackend
    ) -> dict[str, Any]:
        """Execute the form fill on the backend.

        Args:
            backend: The browser backend to use.

        Returns:
            Dict with filled count and submit status.
        """
        await backend.navigate(self.params.url, self.params.wait)

        filled = 0
        for selector, value in self.params.fields.items():
            try:
                await backend.fill(selector, value)
                filled += 1
            except WavexisError:
                pass

        submitted = False
        if self.params.submit:
            try:
                await backend.click(self.params.submit)
                submitted = True
            except WavexisError:
                pass

        return {
            "url": self.params.url,
            "fields_filled": filled,
            "fields_total": len(self.params.fields),
            "submitted": submitted,
        }

execute async

execute(backend: AbstractBackend) -> dict[str, Any]

Execute the form fill on the backend.

Parameters:

Name Type Description Default
backend AbstractBackend

The browser backend to use.

required

Returns:

Type Description
dict[str, Any]

Dict with filled count and submit status.

Source code in wavexis/actions/form.py
async def execute(
    self, backend: AbstractBackend
) -> dict[str, Any]:
    """Execute the form fill on the backend.

    Args:
        backend: The browser backend to use.

    Returns:
        Dict with filled count and submit status.
    """
    await backend.navigate(self.params.url, self.params.wait)

    filled = 0
    for selector, value in self.params.fields.items():
        try:
            await backend.fill(selector, value)
            filled += 1
        except WavexisError:
            pass

    submitted = False
    if self.params.submit:
        try:
            await backend.click(self.params.submit)
            submitted = True
        except WavexisError:
            pass

    return {
        "url": self.params.url,
        "fields_filled": filled,
        "fields_total": len(self.params.fields),
        "submitted": submitted,
    }