Let found elements act on themselves

README.md

@@ -173,6 +173,16 @@ if r.http_status >= 400:
     raise SystemExit(f"{r.url} is an error page, not content")
 ```
 
+`find_element` and `interactive_elements` return elements that act on
+themselves: each is still the dict the browser sent, and every action that
+takes a `backend_node_id` is also a method on it:
+
+```python
+found = page.find_element(role="button", name="Accept")
+if found:
+    found[0].click()        # also fill(value=...), hover(), markdown(), ...
+```
+
 ## Examples
 
 [`examples/`](examples/) scrapes a JavaScript-rendered site that `requests`

lightpanda/__init__.py

@@ -36,19 +36,21 @@
 
 __docformat__ = "google"
 
-from .async_browser import AsyncBrowser, AsyncSession, run_script_async
+from .async_browser import AsyncBrowser, AsyncElement, AsyncSession, run_script_async
 from .webdriver import AsyncWebDriverServer, WebDriverServer
-from .browser import Browser, PageResult, Session, run_script
+from .browser import Browser, Element, PageResult, Session, run_script
 from .cdp import AsyncCDPServer, CDPServer
 from .errors import LightpandaError, ProcessError, ProtocolError, ScriptError, ToolError
 
 __all__ = [
     "Browser",
     "Session",
+    "Element",
     "PageResult",
     "run_script",
     "AsyncBrowser",
     "AsyncSession",
+    "AsyncElement",
     "run_script_async",
     "CDPServer",
     "AsyncCDPServer",

lightpanda/async_browser.py

@@ -17,11 +17,12 @@
 from collections.abc import Sequence
 from concurrent.futures import ThreadPoolExecutor
 
-from .browser import Browser, Session, _generated, run_script
+from .browser import Browser, Element, Session, _generated, run_script
 from .client import _documented
 from .errors import LightpandaError
 
 AsyncSessionMethods = _generated("AsyncSessionMethods")
+AsyncElementMethods = _generated("AsyncElementMethods")
 
 
 async def _run(executor: ThreadPoolExecutor, fn, *args, **kwargs):
@@ -52,7 +53,10 @@ def id(self) -> str:
     async def call(self, tool: str, **kwargs):
         """Invoke a browser tool by name. The generated methods route here.
         Same contract as :meth:`Session.call`, awaitable."""
-        return await _run(self._executor, self._session.call, tool, **kwargs)
+        result = await _run(self._executor, self._session.call, tool, **kwargs)
+        if isinstance(result, list):
+            return [AsyncElement(node, self) if isinstance(node, Element) else node for node in result]
+        return result
 
     def __getattr__(self, attr: str):
         session = self.__dict__.get("_session")
@@ -72,6 +76,34 @@ async def __aexit__(self, *exc):
 
 
 
+@_documented
+class AsyncElement(AsyncElementMethods, dict):
+    """A node :meth:`AsyncSession.find_element` or
+    :meth:`AsyncSession.interactive_elements` found, which acts on itself.
+
+    Same as :class:`Element`, with every method a coroutine to await:
+
+    ```python
+    found = await page.find_element(role="button", name="Accept")
+    if found:
+        await found[0].click()
+    ```
+    """
+
+    def __init__(self, fields: dict, session: AsyncSession):
+        super().__init__(fields)
+        self._session = session
+
+    @property
+    def backend_node_id(self) -> int:
+        """The node's id, as :meth:`AsyncSession.click` and the others take it."""
+        return self["backendNodeId"]
+
+    async def call(self, tool: str, **kwargs):
+        """Invoke a browser tool on this node; see :meth:`AsyncSession.call`."""
+        return await self._session.call(tool, backendNodeId=self.backend_node_id, **kwargs)
+
+
 class AsyncBrowser:
     """A lightpanda browser process, driven from asyncio.
 

lightpanda/browser.py

@@ -73,6 +73,45 @@ def _generated(name: str) -> type:
 
 
 SessionMethods = _generated("SessionMethods")
+ElementMethods = _generated("ElementMethods")
+
+# Tools answering with a list of nodes, which `Session.call` returns as Elements.
+_ELEMENT_TOOLS = {"findElement", "interactiveElements"}
+# Element methods aim at the element itself: no other target, no navigation.
+_ELEMENT_PARAMS = {"selector", "backendNodeId", "url", "timeout"}
+
+
+@_documented
+class Element(ElementMethods, dict):
+    """A node :meth:`Session.find_element` or :meth:`Session.interactive_elements`
+    found, which acts on itself.
+
+    Still the dict the browser sent, so ``element["name"]`` reads its fields,
+    with every action that takes a ``backend_node_id`` as a method aimed at
+    this node:
+
+    ```python
+    found = page.find_element(role="button", name="Accept")
+    if found:
+        found[0].click()
+    ```
+
+    The element belongs to its page: once the page navigates away, its node
+    is gone and the methods fail.
+    """
+
+    def __init__(self, fields: dict, session: Session):
+        super().__init__(fields)
+        self._session = session
+
+    @property
+    def backend_node_id(self) -> int:
+        """The node's id, as :meth:`Session.click` and the others take it."""
+        return self["backendNodeId"]
+
+    def call(self, tool: str, **kwargs):
+        """Invoke a browser tool on this node; see :meth:`Session.call`."""
+        return self._session.call(tool, backendNodeId=self.backend_node_id, **kwargs)
 
 
 @_documented
@@ -167,9 +206,12 @@ def call(self, tool: str, **kwargs):
         if page_state is not None:
             return PageResult(text, page_state)
         try:
-            return json.loads(text)
+            parsed = json.loads(text)
         except ValueError:
             return text
+        if name in _ELEMENT_TOOLS and isinstance(parsed, list):
+            return [Element(node, self) for node in parsed]
+        return parsed
 
     def _resolve(self, attr: str) -> str | None:
         """The tool name behind a snake_case public attribute, if any."""

scripts/generate_methods.py

@@ -8,7 +8,9 @@
 declare an ``outputSchema`` answer with ``structuredContent``, which
 ``Session.call`` returns as a ``PageResult``, so those methods are annotated
 with it and get a ``Returns:`` section listing its attributes from the
-schema's property descriptions; the rest stay ``Any``. Being real code, the methods are
+schema's property descriptions; the rest stay ``Any``. Tools that take a
+``backendNodeId`` also get a method on ``ElementMethods``, aimed at the element
+it is called on. Being real code, the methods are
 visible to IDEs, type checkers, and pdoc alike. Run with a binary available:
 
     uv run --no-project python scripts/generate_methods.py
@@ -25,7 +27,13 @@
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
 from lightpanda import Browser  # noqa: E402
-from lightpanda.browser import _SESSION_TOOLS, PageResult, _snake  # noqa: E402
+from lightpanda.browser import (  # noqa: E402
+    _ELEMENT_PARAMS,
+    _ELEMENT_TOOLS,
+    _SESSION_TOOLS,
+    PageResult,
+    _snake,
+)
 
 PY_TYPES = {"string": "str", "integer": "int", "number": "float", "boolean": "bool", "object": "dict", "array": "list"}
 
@@ -34,16 +42,18 @@
 
 One method per lightpanda browser tool, signatures derived from the tool
 JSON schemas, with tool and parameter names in snake_case (the schemas'
-camelCase names still work through ``Session.call``). ``AsyncSessionMethods``
-is the same surface with ``async def`` signatures.
+camelCase names still work through ``Session.call``). ``ElementMethods``
+holds the tools that take a ``backendNodeId``, aimed at the element itself.
+The ``Async`` classes are the same surfaces with ``async def`` signatures.
 """
 
 from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
-    from .browser import PageResult
+    from .async_browser import AsyncElement
+    from .browser import Element, PageResult
 '''
 
 CLASS_HEADER = '''\
@@ -85,9 +95,11 @@ def docstring(description: str, args: list[tuple[str, str]], returns: str = "")
     return f'        """{indented.lstrip()}\n        """'
 
 
-def method_source(name: str, spec: dict, is_async: bool = False) -> str:
+def method_source(name: str, spec: dict, is_async: bool = False, on_element: bool = False) -> str:
     schema = spec["schema"]
     properties = schema.get("properties", {})
+    if on_element:
+        properties = {prop: v for prop, v in properties.items() if prop not in _ELEMENT_PARAMS}
     required = set(schema.get("required", []))
     args = {prop: _snake(prop) for prop in properties}
     for prop, arg in args.items():
@@ -119,6 +131,8 @@ def method_source(name: str, spec: dict, is_async: bool = False) -> str:
     await_ = "await " if is_async else ""
     output_schema = spec.get("output_schema")
     returns = "PageResult" if output_schema else "Any"
+    if name in _ELEMENT_TOOLS:
+        returns = "list[AsyncElement]" if is_async else "list[Element]"
     lines = [f"    {prefix} {snake}({', '.join(params)}) -> {returns}:"]
     section = returns_section(name, output_schema) if output_schema else ""
     lines.append(docstring(spec["description"], documented, section))
@@ -131,16 +145,22 @@ def main() -> None:
     with Browser() as browser:
         tools = {k: v for k, v in browser.tools.items() if k not in _SESSION_TOOLS}
 
+    on_node = sorted(name for name in tools if "backendNodeId" in tools[name]["schema"].get("properties", {}))
+    # Elements are dicts: a tool must not shadow one of dict's methods.
+    if shadowed := [name for name in on_node if hasattr(dict, _snake(name))]:
+        raise SystemExit(f"tools {shadowed} would shadow dict methods on Element")
     blocks = [HEADER]
     for is_async in (False, True):
-        cls = "AsyncSessionMethods" if is_async else "SessionMethods"
-        blocks.append(CLASS_HEADER.format(cls=cls, prefix="async " if is_async else ""))
+        prefix = "Async" if is_async else ""
+        blocks.append(CLASS_HEADER.format(cls=f"{prefix}SessionMethods", prefix="async " if is_async else ""))
         blocks += [method_source(name, tools[name], is_async) for name in sorted(tools)]
+        blocks.append(CLASS_HEADER.format(cls=f"{prefix}ElementMethods", prefix="async " if is_async else ""))
+        blocks += [method_source(name, tools[name], is_async, on_element=True) for name in on_node]
 
     source = "\n".join(blocks) + "\n"
     compile(source, str(out), "exec")
     out.write_text(source)
-    print(f"wrote {out} ({len(tools)} tools)")
+    print(f"wrote {out} ({len(tools)} tools, {len(on_node)} on elements)")
 
 
 if __name__ == "__main__":

tests/test_async.py

@@ -6,7 +6,13 @@
 
 import pytest
 
-from lightpanda import AsyncBrowser, LightpandaError, ToolError, run_script_async
+from lightpanda import (
+    AsyncBrowser,
+    AsyncElement,
+    LightpandaError,
+    ToolError,
+    run_script_async,
+)
 
 
 async def test_goto_markdown_and_snake_case(abrowser, fixture_url):
@@ -28,6 +34,15 @@ async def visit(path):
     assert [u.rsplit("/", 1)[1] for u in urls] == ["index.html", "other.html", "index.html"]
 
 
+async def test_found_elements_act_on_themselves(abrowser, fixture_url):
+    async with abrowser.session() as page:
+        await page.goto(url=f"{fixture_url}/index.html")
+        found = await page.find_element(role="link", name="Second")
+        assert [type(el) for el in found] == [AsyncElement]
+        assert found[0]["name"] == "Second item"
+        assert (await found[0].click()).url == f"{fixture_url}/other.html"
+
+
 async def test_tool_error_raises(abrowser, fixture_url):
     async with abrowser.session() as page:
         await page.goto(url=f"{fixture_url}/index.html")

tests/test_browser.py

@@ -1,3 +1,4 @@
+import json
 import os
 import signal
 import subprocess
@@ -8,7 +9,7 @@
 import pytest
 
 from conftest import alive
-from lightpanda import Browser, LightpandaError, PageResult, ScriptError, ToolError, run_script, client
+from lightpanda import Browser, Element, LightpandaError, PageResult, ScriptError, ToolError, run_script, client
 
 
 def test_goto_and_markdown(browser, fixture_url):
@@ -45,6 +46,24 @@ def test_page_result_carries_status(browser, fixture_url):
         assert clicked.http_status == 200
 
 
+def test_found_elements_act_on_themselves(browser, fixture_url):
+    with browser.new_session() as page:
+        page.goto(url=f"{fixture_url}/index.html")
+        found = page.find_element(role="link", name="Second")
+        assert [type(el) for el in found] == [Element]
+        link = found[0]
+        # Still the dict the browser sent.
+        assert link["name"] == "Second item"
+        assert json.loads(json.dumps(link)) == dict(link)
+        assert link.backend_node_id == link["backendNodeId"]
+
+        assert link.markdown().strip() == f"[Second item]({fixture_url}/other.html)"
+        assert link.click().url == f"{fixture_url}/other.html"
+
+        page.goto(url=f"{fixture_url}/index.html")
+        assert all(isinstance(el, Element) for el in page.interactive_elements())
+
+
 def test_extract_dict_schema(browser, fixture_url):
     with browser.new_session() as page:
         page.goto(url=f"{fixture_url}/index.html")