Automatisation/browser-cli
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 6 Wiki Activity

refactor: reorganize client transport and extension internals

Browse Source 
- Split client, native, remote, serve, markdown, and SDK internals into focused packages with direct imports.
- Move local and remote transport framing/protocol helpers behind clearer module boundaries.
- Break up the extension injected DOM logic into a separate content dispatch bundle and dedicated content modules.
- Add explicit client handling for passive remote discovery without noisy PQ warnings.
- Keep behavior covered with updated unit, integration, and extension tests.
This commit is contained in:
Daniel Dolezal
2026-06-13 23:31:24 +02:00
parent fd5447cbb9
commit 076914e5b7
88 changed files with 7491 additions and 5228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
browser_cli/sdk/__init__.py
+20
View File
@@ -5,6 +5,7 @@ client (``b.tabs``, ``b.dom``, ``b.session``, ...), mirroring the command groups
in the browser extension.
"""
from browser_cli.sdk.browser_data import CookiesNS, StorageNS
from browser_cli.sdk.decorators import DecoratorsNS
from browser_cli.sdk.dom import DomNS, ExtractNS, PageNS
from browser_cli.sdk.extension import ExtensionNS
from browser_cli.sdk.groups import GroupsNS
@@ -14,6 +15,22 @@ from browser_cli.sdk.session import SessionNS
from browser_cli.sdk.tabs import TabsNS
from browser_cli.sdk.windows import WindowsNS
NAMESPACE_SPECS = (
("nav", NavigationNS),
("tabs", TabsNS),
("groups", GroupsNS),
("windows", WindowsNS),
("dom", DomNS),
("extract", ExtractNS),
("page", PageNS),
("storage", StorageNS),
("cookies", CookiesNS),
("session", SessionNS),
("perf", PerfNS),
("extension", ExtensionNS),
)
NAMESPACE_NAMES = tuple(name for name, _ in NAMESPACE_SPECS)
__all__ = [
"NavigationNS",
"TabsNS",
@@ -27,4 +44,7 @@ __all__ = [
"SessionNS",
"PerfNS",
"ExtensionNS",
"DecoratorsNS",
"NAMESPACE_SPECS",
"NAMESPACE_NAMES",
]
browser_cli/sdk/base.py
+106 -9
View File
@@ -1,19 +1,116 @@
"""Base class for SDK command namespaces.
"""Base helpers for SDK command namespaces.
Each namespace (``b.tabs``, ``b.dom``, ...) is a thin object bound to its
:class:`~browser_cli.BrowserCLI` client. Namespaces hold no state of their own;
they delegate to the client's shared infrastructure (``_cmd``, the multi-browser
helpers, and the ``Tab``/``Group`` factories).
they delegate to the client's shared infrastructure (command dispatch, the
multi-browser helpers, and the ``Tab``/``Group`` factories).
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from collections.abc import Callable
from functools import wraps
from typing import Any, TypeVar
if TYPE_CHECKING:
from browser_cli import BrowserCLI
F = TypeVar("F", bound=Callable)
_MISSING = object()
def _clone_default(value):
if isinstance(value, (dict, list, set)):
return value.copy()
return value
def sdk_command(
name: str,
args: Callable | None = None,
*,
default=_MISSING,
field: str | None = None,
fallback=_MISSING,
mapper: Callable | None = None,
return_result: bool = True,
):
"""Decorate a namespace method as a browser wire command.
This keeps the public method signature/docstring on the normal Python
method, while moving repetitive command-dispatch plumbing into one place.
The wrapped function body is intentionally unused; it exists only for
signature, docs, and type hints.
"""
def decorator(func: F) -> F:
@wraps(func)
def wrapper(self, *method_args, **method_kwargs):
payload = args(self, *method_args, **method_kwargs) if args is not None else {}
result = self.command(name, payload)
if not return_result:
return None
if mapper is not None:
return mapper(self, result, *method_args, **method_kwargs)
if field is not None:
if fallback is _MISSING:
return self.field(result, field, default)
return self.field(result, field, default, fallback=fallback)
if default is not _MISSING and not result:
return _clone_default(default)
return result
wrapper._browser_cli_command = name # type: ignore[attr-defined]
return wrapper # type: ignore[return-value]
return decorator
class Namespace:
"""A group of related SDK methods, bound to a BrowserCLI client."""
"""A group of related SDK methods, bound to a BrowserCLI client."""
def __init__(self, client: "BrowserCLI"):
self._c = client
def __init__(self, client: Any):
self._c = client
def command(self, name: str, args: dict | None = None):
"""Dispatch a browser command through the owning client."""
return self._c.dispatch(name, args)
def tab_from(self, data: dict):
"""Build a bound Tab from a raw command response dict."""
return self._c.tab_from(data)
def group_from(self, data: dict):
"""Build a bound Group from a raw command response dict."""
return self._c.group_from(data)
def tab_from_target(self, data: dict, target):
"""Build a bound Tab for a multi-browser target."""
return self._c.tab_from_target(data, target)
def group_from_target(self, data: dict, target):
"""Build a bound Group for a multi-browser target."""
return self._c.group_from_target(data, target)
def tag_browser(self, item: dict, target):
"""Annotate a raw dict with its browser in multi-browser mode."""
return self._c.tag_browser(item, target)
def multi_list(self, name: str, args: dict | None, mapper: Callable):
"""Run a list command with multi-browser fan-out support."""
return self._c.multi_list(name, args, mapper)
def multi_count(self, name: str, args: dict | None = None):
"""Run a count command with multi-browser fan-out support."""
return self._c.multi_count(name, args)
def apply_tab_filter(self, filter_fn: Callable):
"""Apply a Python-side tab filter using client semantics."""
return self._c.apply_tab_filter(filter_fn)
def toggle_tab(self, name: str, tab_id: int | None):
"""Run a tab toggle command and return the affected tab ID."""
return self._c.toggle_tab(name, tab_id)
def require_tab(self, data, error: str):
"""Convert a tab-like response into a bound Tab or raise a clean error."""
return self._c.require_tab(data, error)
def field(self, result, key, default=None, *, fallback=_MISSING):
"""Read a field from command output using the client's SDK semantics."""
if fallback is _MISSING:
return self._c.field(result, key, default)
return self._c.field(result, key, default, fallback=fallback)
browser_cli/sdk/browser_data.py
+73 -54
View File
@@ -1,66 +1,85 @@
"""Storage and cookies namespaces: ``b.storage.*``, ``b.cookies.*``."""
from __future__ import annotations
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.base import Namespace, sdk_command
class StorageNS(Namespace):
"""Read and write localStorage / sessionStorage."""
"""Read and write localStorage / sessionStorage."""
def get(
self,
key: str | None = None,
*,
type: str = "local",
tab_id: int | None = None,
) -> str | dict | None:
"""Get a localStorage/sessionStorage entry (or all entries if key omitted)."""
return self._c._cmd("storage.get", {"key": key, "type": type, "tabId": tab_id})
@sdk_command("storage.get", lambda self, key=None, *, type="local", tab_id=None: {
"key": key,
"type": type,
"tabId": tab_id,
})
def get(
self,
key: str | None = None,
*,
type: str = "local",
tab_id: int | None = None,
) -> str | dict | None:
"""Get a localStorage/sessionStorage entry (or all entries if key omitted)."""
def set(
self,
key: str,
value: str,
*,
type: str = "local",
tab_id: int | None = None,
) -> None:
"""Set a localStorage/sessionStorage entry."""
self._c._cmd("storage.set", {"key": key, "value": value, "type": type, "tabId": tab_id})
@sdk_command("storage.set", lambda self, key, value, *, type="local", tab_id=None: {
"key": key,
"value": value,
"type": type,
"tabId": tab_id,
})
def set(
self,
key: str,
value: str,
*,
type: str = "local",
tab_id: int | None = None,
) -> None:
"""Set a localStorage/sessionStorage entry."""
class CookiesNS(Namespace):
"""List, get, and set cookies."""
"""List, get, and set cookies."""
def list(
self,
*,
url: str | None = None,
domain: str | None = None,
name: str | None = None,
) -> list[dict]:
"""List cookies, optionally filtered by url, domain, or name."""
return self._c._cmd("cookies.list", {"url": url, "domain": domain, "name": name}) or []
@sdk_command("cookies.list", lambda self, *, url=None, domain=None, name=None: {
"url": url,
"domain": domain,
"name": name,
}, default=[])
def list(
self,
*,
url: str | None = None,
domain: str | None = None,
name: str | None = None,
) -> list[dict]:
"""List cookies, optionally filtered by url, domain, or name."""
def get(self, url: str, name: str) -> dict | None:
"""Get a single cookie by url and name."""
return self._c._cmd("cookies.get", {"url": url, "name": name})
@sdk_command("cookies.get", lambda self, url, name: {"url": url, "name": name})
def get(self, url: str, name: str) -> dict | None:
"""Get a single cookie by url and name."""
def set(
self,
url: str,
name: str,
value: str,
*,
domain: str | None = None,
path: str | None = None,
secure: bool | None = None,
http_only: bool | None = None,
expiration_date: float | None = None,
same_site: str | None = None,
) -> dict:
"""Set a cookie. Returns the created cookie dict."""
return self._c._cmd("cookies.set", {
"url": url, "name": name, "value": value,
"domain": domain, "path": path,
"secure": secure, "httpOnly": http_only,
"expirationDate": expiration_date, "sameSite": same_site,
})
@sdk_command("cookies.set", lambda self, url, name, value, *, domain=None, path=None, secure=None,
http_only=None, expiration_date=None, same_site=None: {
"url": url,
"name": name,
"value": value,
"domain": domain,
"path": path,
"secure": secure,
"httpOnly": http_only,
"expirationDate": expiration_date,
"sameSite": same_site,
})
def set(
self,
url: str,
name: str,
value: str,
*,
domain: str | None = None,
path: str | None = None,
secure: bool | None = None,
http_only: bool | None = None,
expiration_date: float | None = None,
same_site: str | None = None,
) -> dict:
"""Set a cookie. Returns the created cookie dict."""
browser_cli/sdk/decorators.py
+107
View File
@@ -0,0 +1,107 @@
"""Synchronous workflow decorator namespace for the Python SDK."""
from __future__ import annotations
import asyncio
import functools
import inspect
from collections.abc import Callable
from typing import TypeVar
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.workflow_decorators import WorkflowDecoratorsMixin, _NO_INJECT
F = TypeVar("F", bound=Callable)
class DecoratorsNS(WorkflowDecoratorsMixin, Namespace):
"""Workflow decorators bound to a :class:`browser_cli.BrowserCLI` client.
The normal SDK is synchronous, but these decorators also work on ``async def``
functions: browser operations run via ``asyncio.to_thread`` so the event loop
is not blocked while waiting for the browser response.
"""
def _run(self, func: Callable, *args, **kwargs):
if inspect.iscoroutinefunction(func):
raise TypeError("sync BrowserCLI decorators cannot call async browser methods")
return func(*args, **kwargs)
def _call_wrapped(self, func: Callable, *args, **kwargs):
if inspect.iscoroutinefunction(func):
async def run_async():
return await func(*args, **kwargs)
return run_async()
return func(*args, **kwargs)
def _value_decorator(
self,
func: F | None,
get_value: Callable,
*,
keyword: str | None | object = "tab",
cleanup: Callable | None = None,
):
def decorator(fn: F) -> F:
if inspect.iscoroutinefunction(fn):
@functools.wraps(fn)
async def async_wrapper(*args, **kwargs):
value = await asyncio.to_thread(get_value)
try:
extra_args = ()
if keyword is not _NO_INJECT:
extra_args, kwargs = self._inject(kwargs, keyword, value)
return await fn(*extra_args, *args, **kwargs)
finally:
if cleanup is not None:
await asyncio.to_thread(cleanup, value)
return async_wrapper # type: ignore[return-value]
return WorkflowDecoratorsMixin._value_decorator(
self, fn, get_value, keyword=keyword, cleanup=cleanup
)
return decorator(func) if func is not None else decorator
def performance_profile(self, profile: str, *, restore: bool = True):
def decorator(fn: F) -> F:
if inspect.iscoroutinefunction(fn):
@functools.wraps(fn)
async def async_wrapper(*args, **kwargs):
previous = None
if restore:
previous = (await asyncio.to_thread(self._c.perf.status)).get("performanceProfile")
await asyncio.to_thread(self._c.perf.set_profile, profile)
try:
return await fn(*args, **kwargs)
finally:
if previous:
await asyncio.to_thread(self._c.perf.set_profile, previous)
return async_wrapper # type: ignore[return-value]
return WorkflowDecoratorsMixin.performance_profile(self, profile, restore=restore)(fn)
return decorator
def retry(
self,
*,
times: int = 3,
delay: float = 0.0,
exceptions: tuple[type[BaseException], ...] = (Exception,),
):
attempts = max(1, times)
def decorator(fn: F) -> F:
if inspect.iscoroutinefunction(fn):
@functools.wraps(fn)
async def async_wrapper(*args, **kwargs):
last_error = None
for attempt in range(attempts):
try:
return await fn(*args, **kwargs)
except exceptions as exc:
last_error = exc
if attempt == attempts - 1:
raise
if delay > 0:
await asyncio.sleep(delay)
raise last_error # type: ignore[misc]
return async_wrapper # type: ignore[return-value]
return WorkflowDecoratorsMixin.retry(self, times=times, delay=delay, exceptions=exceptions)(fn)
return decorator
browser_cli/sdk/dom.py
+132 -113
View File
@@ -1,150 +1,169 @@
"""DOM, content-extraction, and page-info namespaces: ``b.dom.*``, ``b.extract.*``, ``b.page.*``."""
from __future__ import annotations
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.base import Namespace, sdk_command
def _selector_args(self, selector):
return {"selector": selector}
def _selector_value_args(self, selector, value):
return {"selector": selector, "value": value}
def _extract_markdown(self, result, *args, **kwargs) -> str:
from browser_cli.markdown import render_markdown
return render_markdown(result)
class DomNS(Namespace):
"""Query and drive page elements in the active (or specified) tab."""
"""Query and drive page elements in the active (or specified) tab."""
def query(self, selector: str) -> list[dict]:
return self._c._cmd("dom.query", {"selector": selector}) or []
@sdk_command("dom.query", _selector_args, default=[])
def query(self, selector: str) -> list[dict]:
"""Return elements matching a CSS selector."""
def click(self, selector: str) -> None:
self._c._cmd("dom.click", {"selector": selector})
@sdk_command("dom.click", _selector_args, return_result=False)
def click(self, selector: str) -> None:
"""Click the first element matching a CSS selector."""
def type(self, selector: str, text: str) -> None:
self._c._cmd("dom.type", {"selector": selector, "text": text})
@sdk_command("dom.type", lambda self, selector, text: {"selector": selector, "text": text}, return_result=False)
def type(self, selector: str, text: str) -> None:
"""Type text into the first matching element."""
def attr(self, selector: str, attr: str) -> list[str]:
return self._c._cmd("dom.attr", {"selector": selector, "attr": attr}) or []
@sdk_command("dom.attr", lambda self, selector, attr: {"selector": selector, "attr": attr}, default=[])
def attr(self, selector: str, attr: str) -> list[str]:
"""Return an attribute from all matching elements."""
def text(self, selector: str) -> list[str]:
return self._c._cmd("dom.text", {"selector": selector}) or []
@sdk_command("dom.text", _selector_args, default=[])
def text(self, selector: str) -> list[str]:
"""Return text from all matching elements."""
def exists(self, selector: str) -> bool:
return self._c._cmd("dom.exists", {"selector": selector}) or False
@sdk_command("dom.exists", _selector_args, default=False)
def exists(self, selector: str) -> bool:
"""Return whether a selector exists."""
def scroll(self, selector: str | None = None, *, x: int | None = None, y: int | None = None) -> None:
"""Scroll to a CSS selector or to pixel coordinates."""
self._c._cmd("dom.scroll", {"selector": selector, "x": x, "y": y})
@sdk_command("dom.scroll", lambda self, selector=None, *, x=None, y=None: {"selector": selector, "x": x, "y": y}, return_result=False)
def scroll(self, selector: str | None = None, *, x: int | None = None, y: int | None = None) -> None:
"""Scroll to a CSS selector or to pixel coordinates."""
def select(self, selector: str, value: str) -> None:
"""Set the value of a <select> element."""
self._c._cmd("dom.select", {"selector": selector, "value": value})
@sdk_command("dom.select", _selector_value_args, return_result=False)
def select(self, selector: str, value: str) -> None:
"""Set the value of a <select> element."""
def eval(self, code: str, tab_id: int | None = None):
"""Evaluate JavaScript in the page's main world and return the result."""
return self._c._cmd("dom.eval", {"code": code, "tabId": tab_id})
@sdk_command("dom.eval", lambda self, code, tab_id=None: {"code": code, "tabId": tab_id})
def eval(self, code: str, tab_id: int | None = None):
"""Evaluate JavaScript in the page's main world and return the result."""
def key(self, key: str, selector: str | None = None) -> None:
"""Dispatch a keyboard event. key examples: 'Enter', 'Tab', 'Escape', 'ArrowDown'."""
self._c._cmd("dom.key", {"key": key, "selector": selector})
@sdk_command("dom.key", lambda self, key, selector=None: {"key": key, "selector": selector}, return_result=False)
def key(self, key: str, selector: str | None = None) -> None:
"""Dispatch a keyboard event. key examples: 'Enter', 'Tab', 'Escape', 'ArrowDown'."""
def hover(self, selector: str) -> None:
"""Dispatch mouseover/mouseenter on an element."""
self._c._cmd("dom.hover", {"selector": selector})
@sdk_command("dom.hover", _selector_args, return_result=False)
def hover(self, selector: str) -> None:
"""Dispatch mouseover/mouseenter on an element."""
def check(self, selector: str) -> None:
"""Check a checkbox."""
self._c._cmd("dom.check", {"selector": selector})
@sdk_command("dom.check", _selector_args, return_result=False)
def check(self, selector: str) -> None:
"""Check a checkbox."""
def uncheck(self, selector: str) -> None:
"""Uncheck a checkbox."""
self._c._cmd("dom.uncheck", {"selector": selector})
@sdk_command("dom.uncheck", _selector_args, return_result=False)
def uncheck(self, selector: str) -> None:
"""Uncheck a checkbox."""
def clear(self, selector: str) -> None:
"""Clear the value of an input element."""
self._c._cmd("dom.clear", {"selector": selector})
@sdk_command("dom.clear", _selector_args, return_result=False)
def clear(self, selector: str) -> None:
"""Clear the value of an input element."""
def focus(self, selector: str) -> None:
"""Focus an element."""
self._c._cmd("dom.focus", {"selector": selector})
@sdk_command("dom.focus", _selector_args, return_result=False)
def focus(self, selector: str) -> None:
"""Focus an element."""
def submit(self, selector: str) -> None:
"""Submit the form containing the matched element."""
self._c._cmd("dom.submit", {"selector": selector})
@sdk_command("dom.submit", _selector_args, return_result=False)
def submit(self, selector: str) -> None:
"""Submit the form containing the matched element."""
def poll(
self,
selector: str,
pattern: str,
*,
attr: str | None = None,
timeout: float = 30.0,
interval: float = 0.5,
tab_id: int | None = None,
) -> dict:
"""Poll selector's text/value until it matches regex pattern.
def poll(
self,
selector: str,
pattern: str,
*,
attr: str | None = None,
timeout: float = 30.0,
interval: float = 0.5,
tab_id: int | None = None,
) -> dict:
"""Poll selector's text/value until it matches regex pattern.
Returns ``{"selector": ..., "value": ..., "pattern": ...}`` when matched.
"""
return self._c._cmd("dom.poll", {
"selector": selector,
"pattern": pattern,
"attr": attr,
"timeout": int(timeout * 1000),
"interval": int(interval * 1000),
"tabId": tab_id,
})
Returns ``{"selector": ..., "value": ..., "pattern": ...}`` when matched.
"""
return self.command("dom.poll", {
"selector": selector,
"pattern": pattern,
"attr": attr,
"timeout": int(timeout * 1000),
"interval": int(interval * 1000),
"tabId": tab_id,
})
def wait_for(
self,
selector: str,
*,
timeout: float = 10.0,
visible: bool = False,
hidden: bool = False,
tab_id: int | None = None,
) -> dict:
"""Wait until a CSS selector appears (or disappears) in the DOM.
def wait_for(
self,
selector: str,
*,
timeout: float = 10.0,
visible: bool = False,
hidden: bool = False,
tab_id: int | None = None,
) -> dict:
"""Wait until a CSS selector appears (or disappears) in the DOM.
Args:
selector: CSS selector to watch.
timeout: Max seconds to wait before raising ``RuntimeError``.
visible: Wait until the element has non-zero dimensions.
hidden: Wait until the element is absent or has ``offsetParent == null``.
tab_id: Tab to watch. Defaults to the active tab.
"""
return self._c._cmd("dom.wait_for", {
"selector": selector,
"timeout": int(timeout * 1000),
"visible": visible,
"hidden": hidden,
"tabId": tab_id,
})
Args:
selector: CSS selector to watch.
timeout: Max seconds to wait before raising ``RuntimeError``.
visible: Wait until the element has non-zero dimensions.
hidden: Wait until the element is absent or has ``offsetParent == null``.
tab_id: Tab to watch. Defaults to the active tab.
"""
return self.command("dom.wait_for", {
"selector": selector,
"timeout": int(timeout * 1000),
"visible": visible,
"hidden": hidden,
"tabId": tab_id,
})
class ExtractNS(Namespace):
"""Extract structured content from the active tab."""
"""Extract structured content from the active tab."""
def links(self) -> list[dict]:
return self._c._cmd("extract.links", {}) or []
@sdk_command("extract.links", default=[])
def links(self) -> list[dict]:
"""Return links from the active tab."""
def images(self) -> list[dict]:
return self._c._cmd("extract.images", {}) or []
@sdk_command("extract.images", default=[])
def images(self) -> list[dict]:
"""Return images from the active tab."""
def text(self) -> str:
return self._c._cmd("extract.text", {}) or ""
@sdk_command("extract.text", default="")
def text(self) -> str:
"""Return plain text from the active tab."""
def json(self, selector: str):
return self._c._cmd("extract.json", {"selector": selector})
@sdk_command("extract.json", lambda self, selector: {"selector": selector})
def json(self, selector: str):
"""Extract JSON-like structured data from a selector."""
def html(self) -> str:
"""Return the full HTML source of the active tab."""
return self._c._cmd("extract.html", {}) or ""
@sdk_command("extract.html", default="")
def html(self) -> str:
"""Return the full HTML source of the active tab."""
def markdown(self, selector: str | None = None) -> str:
"""Extract the page's main content as clean Markdown.
@sdk_command("extract.markdown", lambda self, selector=None: {"selector": selector}, mapper=_extract_markdown)
def markdown(self, selector: str | None = None) -> str:
"""Extract the page's main content as clean Markdown.
The extractor may return either Markdown or raw HTML; both are
normalized to Markdown here so SDK and CLI callers get identical output.
"""
from browser_cli.markdown import render_markdown
return render_markdown(self._c._cmd("extract.markdown", {"selector": selector}))
The extractor may return either Markdown or raw HTML; both are
normalized to Markdown here so SDK and CLI callers get identical output.
"""
class PageNS(Namespace):
"""Inspect the active page."""
"""Inspect the active page."""
def info(self) -> dict:
"""Return title, URL, readyState, lang, and meta tags of the active tab."""
return self._c._cmd("page.info", {}) or {}
@sdk_command("page.info", default={})
def info(self) -> dict:
"""Return title, URL, readyState, lang, and meta tags of the active tab."""
browser_cli/sdk/extension.py
+9 -9
View File
@@ -1,16 +1,16 @@
"""Extension-control namespace: ``b.extension.*``."""
from __future__ import annotations
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.base import Namespace, sdk_command
class ExtensionNS(Namespace):
"""Control the browser-cli extension itself."""
"""Control the browser-cli extension itself."""
def reload(self) -> None:
"""Reload the browser-cli extension service worker.
@sdk_command("extension.reload")
def reload(self) -> None:
"""Reload the browser-cli extension service worker.
Schedules a ``chrome.runtime.reload()`` inside the extension and returns
immediately. The extension restarts ~200 ms later and reconnects via the
keepalive alarm within ~25 seconds.
"""
self._c._cmd("extension.reload", {})
Schedules a ``chrome.runtime.reload()`` inside the extension and returns
immediately. The extension restarts ~200 ms later and reconnects via the
keepalive alarm within ~25 seconds.
"""
browser_cli/sdk/factories.py
+22 -13
View File
@@ -7,8 +7,13 @@ client targeting the browser it came from, so ``tab.close()`` routes correctly.
"""
from __future__ import annotations
from typing import Any, Protocol, cast
from browser_cli.models import Group, Tab
class _FactoryClient(Protocol):
_key: str | None
class FactoryMixin:
"""Turn raw response dicts into bound ``Tab``/``Group`` objects.
@@ -16,7 +21,7 @@ class FactoryMixin:
``_browser``/``_remote``/``_key`` and being constructible via ``type(self)``.
"""
def _make_tab(
def tab_from(
self,
data: dict,
*,
@@ -34,20 +39,22 @@ class FactoryMixin:
group_id=data.get("groupId") or None,
browser=browser_name,
)
tab._browser = self if browser_profile is None else type(self)(
client = cast(_FactoryClient, self)
tab._browser = self if browser_profile is None else cast(Any, type(self))(
browser=browser_profile,
remote=browser_remote,
key=self._key,
key=client._key,
_command_sender=getattr(self, "_command_sender", None),
)
return tab
def _require_tab(self, data, error: str) -> Tab:
def require_tab_response(self, data, error: str) -> Tab:
"""Build a bound Tab from a tab-shaped response, or raise ``RuntimeError(error)``."""
if not isinstance(data, dict) or "id" not in data:
raise RuntimeError(error)
return self._make_tab(data)
return self.tab_from(data)
def _make_group(
def group_from(
self,
data: dict,
*,
@@ -63,25 +70,27 @@ class FactoryMixin:
tab_count=data.get("tabCount", 0),
browser=browser_name,
)
group._browser = self if browser_profile is None else type(self)(
client = cast(_FactoryClient, self)
group._browser = self if browser_profile is None else cast(Any, type(self))(
browser=browser_profile,
remote=browser_remote,
key=self._key,
key=client._key,
_command_sender=getattr(self, "_command_sender", None),
)
return group
def _make_tab_for(self, data: dict, target) -> Tab:
def tab_from_target(self, data: dict, target) -> Tab:
"""Build a Tab, tagging it with *target* in multi-browser mode (``None`` = local)."""
return self._make_tab(
return self.tab_from(
data,
browser_profile=target.profile if target else None,
browser_name=target.display_name if target else None,
browser_remote=target.remote if target else None,
)
def _make_group_for(self, data: dict, target) -> Group:
def group_from_target(self, data: dict, target) -> Group:
"""Build a Group, tagging it with *target* in multi-browser mode (``None`` = local)."""
return self._make_group(
return self.group_from(
data,
browser_profile=target.profile if target else None,
browser_name=target.display_name if target else None,
@@ -89,6 +98,6 @@ class FactoryMixin:
)
@staticmethod
def _tag_browser(item: dict, target) -> dict:
def tag_browser(item: dict, target) -> dict:
"""Return *item* as-is locally, or with a ``browser`` key in multi-browser mode."""
return item if target is None else {**item, "browser": target.display_name}
browser_cli/sdk/groups.py
+35 -40
View File
@@ -1,56 +1,51 @@
"""Tab groups namespace: ``b.groups.*``."""
from __future__ import annotations
from typing import TYPE_CHECKING
from browser_cli.models import Group, Tab
from browser_cli.models import BrowserCounts, Group, Tab
from browser_cli.sdk.base import Namespace
if TYPE_CHECKING:
from browser_cli import BrowserCounts
class GroupsNS(Namespace):
"""List, create, query, and modify tab groups."""
"""List, create, query, and modify tab groups."""
def list(self) -> list[Group]:
"""Return all tab groups.
def list(self) -> list[Group]:
"""Return all tab groups.
When multiple browsers are active and no browser was specified, each Group
includes ``group.browser`` naming its source browser.
"""
return self._c._multi_list("group.list", {}, self._c._make_group_for)
When multiple browsers are active and no browser was specified, each Group
includes ``group.browser`` naming its source browser.
"""
return self.multi_list("group.list", {}, self.group_from_target)
def count(self) -> "int | BrowserCounts":
"""Return the number of tab groups.
def count(self) -> "int | BrowserCounts":
"""Return the number of tab groups.
Returns ``BrowserCounts`` in implicit multi-browser mode.
"""
return self._c._multi_count("group.count", {})
Returns ``BrowserCounts`` in implicit multi-browser mode.
"""
return self.multi_count("group.count", {})
def query(self, search: str) -> list[Group]:
"""Search groups by name."""
return [self._c._make_group(g) for g in (self._c._cmd("group.query", {"search": search}) or [])]
def query(self, search: str) -> list[Group]:
"""Search groups by name."""
return [self.group_from(g) for g in (self.command("group.query", {"search": search}) or [])]
def create(self, name: str) -> Group:
"""Create a new tab group with *name*. Returns the created Group."""
data = self._c._cmd("group.open", {"name": name})
if isinstance(data, dict):
return self._c._make_group(data)
return Group(id=data, title=name, color="", collapsed=False, tab_count=0)
def create(self, name: str) -> Group:
"""Create a new tab group with *name*. Returns the created Group."""
data = self.command("group.open", {"name": name})
if isinstance(data, dict):
return self.group_from(data)
return Group(id=data, title=name, color="", collapsed=False, tab_count=0)
def tabs(self, group_id: int) -> list[Tab]:
"""Return all tabs inside a group."""
return [self._c._make_tab(t) for t in (self._c._cmd("group.tabs", {"groupId": group_id}) or [])]
def tabs(self, group_id: int) -> list[Tab]:
"""Return all tabs inside a group."""
return [self.tab_from(t) for t in (self.command("group.tabs", {"groupId": group_id}) or [])]
def add_tab(self, group: str | int, url: str | None = None) -> int | None:
"""Open a new tab (optionally at URL) inside a group. Returns the new tab ID."""
result = self._c._cmd("group.add_tab", {"group": str(group), "url": url})
return self._c._field(result, "tabId", fallback=result)
def add_tab(self, group: str | int, url: str | None = None) -> int | None:
"""Open a new tab (optionally at URL) inside a group. Returns the new tab ID."""
result = self.command("group.add_tab", {"group": str(group), "url": url})
return self.field(result, "tabId", fallback=result)
def move(self, group: str | int, *, forward: bool = False, backward: bool = False) -> dict | None:
"""Move a tab group forward or backward. Returns the raw move result."""
return self._c._cmd("group.move", {"group": str(group), "forward": forward, "backward": backward})
def move(self, group: str | int, *, forward: bool = False, backward: bool = False) -> dict | None:
"""Move a tab group forward or backward. Returns the raw move result."""
return self.command("group.move", {"group": str(group), "forward": forward, "backward": backward})
def close(self, group_id: int, *, gentle_mode: str = "auto") -> None:
"""Ungroup (and close) a tab group by ID."""
self._c._cmd("group.close", {"groupId": group_id, "gentleMode": gentle_mode})
def close(self, group_id: int, *, gentle_mode: str = "auto") -> None:
"""Ungroup (and close) a tab group by ID."""
self.command("group.close", {"groupId": group_id, "gentleMode": gentle_mode})
browser_cli/sdk/navigation.py
+58 -48
View File
@@ -2,62 +2,72 @@
from __future__ import annotations
from browser_cli.models import Tab
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.base import Namespace, sdk_command
def _open_args(self, url, *, background=False, window=None, group=None):
return {"url": url, "background": background, "window": window, "group": group}
def _tab_args(self, tab_id=None):
return {"tabId": tab_id}
class NavigationNS(Namespace):
"""Open URLs, navigate history, and focus tabs."""
"""Open URLs, navigate history, and focus tabs."""
def open(self, url: str, *, background: bool = False, window: str | None = None, group: str | None = None) -> None:
"""Open *url* in a new tab."""
self._c._cmd("navigate.open", {"url": url, "background": background, "window": window, "group": group})
@sdk_command("navigate.open", _open_args)
def open(self, url: str, *, background: bool = False, window: str | None = None, group: str | None = None) -> None:
"""Open *url* in a new tab."""
def open_wait(
self,
url: str,
*,
timeout: float = 30.0,
background: bool = False,
window: str | None = None,
group: str | None = None,
) -> Tab:
"""Open *url* in a new tab and block until fully loaded. Returns the Tab."""
return self._c._require_tab(
self._c._cmd("navigate.open_wait", {
"url": url, "timeout": int(timeout * 1000),
"background": background, "window": window, "group": group,
}),
"navigate.open_wait returned unexpected data",
)
def open_wait(
self,
url: str,
*,
timeout: float = 30.0,
background: bool = False,
window: str | None = None,
group: str | None = None,
) -> Tab:
"""Open *url* in a new tab and block until fully loaded. Returns the Tab."""
return self.require_tab(
self.command("navigate.open_wait", {
"url": url, "timeout": int(timeout * 1000),
"background": background, "window": window, "group": group,
}),
"navigate.open_wait returned unexpected data",
)
def reload(self, tab_id: int | None = None) -> None:
self._c._cmd("navigate.reload", {"tabId": tab_id})
@sdk_command("navigate.reload", _tab_args)
def reload(self, tab_id: int | None = None) -> None:
"""Reload the active tab or a specific tab."""
def hard_reload(self, tab_id: int | None = None) -> None:
self._c._cmd("navigate.hard_reload", {"tabId": tab_id})
@sdk_command("navigate.hard_reload", _tab_args)
def hard_reload(self, tab_id: int | None = None) -> None:
"""Hard-reload the active tab or a specific tab."""
def back(self, tab_id: int | None = None) -> None:
self._c._cmd("navigate.back", {"tabId": tab_id})
@sdk_command("navigate.back", _tab_args)
def back(self, tab_id: int | None = None) -> None:
"""Navigate back in the active tab or a specific tab."""
def forward(self, tab_id: int | None = None) -> None:
self._c._cmd("navigate.forward", {"tabId": tab_id})
@sdk_command("navigate.forward", _tab_args)
def forward(self, tab_id: int | None = None) -> None:
"""Navigate forward in the active tab or a specific tab."""
def focus(self, pattern: str) -> dict | None:
"""Focus the first tab whose URL matches *pattern*. Returns the matched tab info, if any."""
return self._c._cmd("navigate.focus", {"pattern": pattern})
@sdk_command("navigate.focus", lambda self, pattern: {"pattern": pattern})
def focus(self, pattern: str) -> dict | None:
"""Focus the first tab whose URL matches *pattern*. Returns the matched tab info, if any."""
def to(self, tab_id: int, url: str) -> None:
"""Navigate a specific tab to *url* in place."""
self._c._cmd("navigate.to", {"tabId": tab_id, "url": url})
@sdk_command("navigate.to", lambda self, tab_id, url: {"tabId": tab_id, "url": url})
def to(self, tab_id: int, url: str) -> None:
"""Navigate a specific tab to *url* in place."""
def search(
self, engine: str, query: str, *,
background: bool = False, window: str | None = None, group: str | None = None,
) -> None:
"""Open a search query in the given engine (e.g. 'google', 'youtube', 'ddg')."""
from urllib.parse import quote_plus
from browser_cli.commands.search import ENGINES
template = ENGINES.get(engine)
if template is None:
raise ValueError(f"Unknown search engine '{engine}'. Available: {', '.join(ENGINES)}")
url = template.format(query=quote_plus(query))
self._c._cmd("navigate.open", {"url": url, "background": background, "window": window, "group": group})
def search(
self, engine: str, query: str, *,
background: bool = False, window: str | None = None, group: str | None = None,
) -> None:
"""Open a search query in the given engine (e.g. 'google', 'youtube', 'ddg')."""
from urllib.parse import quote_plus
from browser_cli.commands.search import ENGINES
template = ENGINES.get(engine)
if template is None:
raise ValueError(f"Unknown search engine '{engine}'. Available: {', '.join(ENGINES)}")
url = template.format(query=quote_plus(query))
self.command("navigate.open", {"url": url, "background": background, "window": window, "group": group})
browser_cli/sdk/perf.py
+14 -10
View File
@@ -1,19 +1,23 @@
"""Performance and background-jobs namespace: ``b.perf.*``."""
from __future__ import annotations
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.base import Namespace, sdk_command
class PerfNS(Namespace):
"""Inspect the performance profile and manage background jobs."""
"""Inspect the performance profile and manage background jobs."""
def status(self) -> dict:
return self._c._cmd("perf.status", {}) or {}
@sdk_command("perf.status", default={})
def status(self) -> dict:
"""Return current performance profile, throttle info, and running jobs."""
def set_profile(self, profile: str) -> dict:
return self._c._cmd("perf.set_profile", {"profile": profile}) or {}
@sdk_command("perf.set_profile", lambda self, profile: {"profile": profile}, default={})
def set_profile(self, profile: str) -> dict:
"""Set the global extension performance profile."""
def job_status(self, job_id: str) -> dict:
return self._c._cmd("jobs.status", {"jobId": job_id}) or {}
@sdk_command("jobs.status", lambda self, job_id: {"jobId": job_id}, default={})
def job_status(self, job_id: str) -> dict:
"""Return status/progress for a background job."""
def job_cancel(self, job_id: str) -> dict:
return self._c._cmd("jobs.cancel", {"jobId": job_id}) or {}
@sdk_command("jobs.cancel", lambda self, job_id: {"jobId": job_id}, default={})
def job_cancel(self, job_id: str) -> dict:
"""Request cancellation for a background job."""
browser_cli/sdk/routing.py
+118 -92
View File
@@ -7,117 +7,143 @@ helpers; single-browser mode falls straight through to ``_cmd``.
"""
from __future__ import annotations
import importlib
import sys
from collections.abc import Callable, Iterable
from typing import TYPE_CHECKING, Protocol, cast
import browser_cli as _pkg
from browser_cli.client import BrowserNotConnected
from browser_cli.client import BrowserTarget
from browser_cli.errors import BrowserNotConnected
from browser_cli.models import BrowserCounts, Tab
if TYPE_CHECKING:
from browser_cli.sdk.tabs import TabsNS
class _RoutingClient(Protocol):
_browser: str | None
_remote: str | None
_key: str | None
tabs: "TabsNS"
def dispatch(self, command: str, args: dict | None = None): ...
# send_command / active_browser_targets / remote_browser_targets are resolved
# through the ``browser_cli`` package namespace (``_pkg``) at call time, not bound
# here at import, so tests patching ``browser_cli.send_command`` still take effect.
# through the ``browser_cli`` package namespace at call time, not bound here at
# import, so tests patching ``browser_cli.send_command`` still take effect —
# without a module-level import cycle back into browser_cli.__init__.
_UNSET = object()
def _browser_cli_package():
return sys.modules.get("browser_cli") or importlib.import_module("browser_cli")
class RoutingMixin:
"""Fan-out + aggregation across active browsers, mixed into ``BrowserCLI``.
"""Fan-out + aggregation across active browsers, mixed into ``BrowserCLI``.
Relies on the client exposing ``_browser``/``_remote``/``_key``, ``_cmd``,
and a ``tabs`` namespace.
"""
Relies on the client exposing ``_browser``/``_remote``/``_key``, ``_cmd``,
and a ``tabs`` namespace.
"""
def _multi_browser_targets(self):
if self._browser is not None:
return []
if self._remote:
targets = _pkg.remote_browser_targets(self._remote, key=self._key)
@property
def _client(self) -> _RoutingClient:
return cast(_RoutingClient, cast(object, self))
def _multi_browser_targets(self) -> list[BrowserTarget]:
client = self._client
package = _browser_cli_package()
if client._browser is not None:
return []
if client._remote:
targets = package.remote_browser_targets(client._remote, key=client._key)
else:
targets = package.active_browser_targets()
if len(targets) <= 1 and not any(target.remote for target in targets):
return []
return targets
def _collect_multi_browser(self, command: str, args: dict | None = None):
results = []
targets = self._multi_browser_targets()
for target in targets:
try:
if target.remote:
data = _browser_cli_package().send_command(
command, args, profile=target.profile, remote=target.remote, key=self._client._key
)
else:
targets = _pkg.active_browser_targets()
if len(targets) <= 1 and not any(target.remote for target in targets):
return []
return targets
data = _browser_cli_package().send_command(command, args, profile=target.profile)
except (BrowserNotConnected, RuntimeError):
continue
results.append((target, data))
if results:
return results
if targets:
raise BrowserNotConnected(
"Cannot resolve a browser socket automatically.\n"
"Make sure the browser is running with the browser-cli extension enabled,\n"
"or pass --browser <alias> / set BROWSER_CLI_PROFILE to a known alias."
)
return []
def _collect_multi_browser(self, command: str, args: dict | None = None):
results = []
targets = self._multi_browser_targets()
for target in targets:
try:
if target.remote:
data = _pkg.send_command(command, args, profile=target.profile, remote=target.remote, key=self._key)
else:
data = _pkg.send_command(command, args, profile=target.profile)
except (BrowserNotConnected, RuntimeError):
continue
results.append((target, data))
if results:
return results
if targets:
raise BrowserNotConnected(
"Cannot resolve a browser socket automatically.\n"
"Make sure the browser is running with the browser-cli extension enabled,\n"
"or pass --browser <alias> / set BROWSER_CLI_PROFILE to a known alias."
)
return []
@staticmethod
def _field(result, key, default=None, *, fallback=_UNSET):
"""Pull *key* out of a dict response, with a non-dict fallback.
@staticmethod
def _field(result, key, default=None, *, fallback=_UNSET):
"""Pull *key* out of a dict response, with a non-dict fallback.
Returns ``result[key]`` (or *default*) when *result* is a dict. When it
is not a dict, returns *fallback* if given, else *default*.
"""
if isinstance(result, dict):
return result.get(key, default)
return default if fallback is _UNSET else fallback
Returns ``result[key]`` (or *default*) when *result* is a dict. When it
is not a dict, returns *fallback* if given, else *default*.
"""
if isinstance(result, dict):
return result.get(key, default)
return default if fallback is _UNSET else fallback
def toggle_tab(self, command: str, tab_id: int | None) -> int:
"""Run a tab toggle command (mute/pin/...) and return the target tab ID."""
result = self._client.dispatch(command, {"tabId": tab_id})
return self._field(result, "tabId", tab_id, fallback=int(tab_id or 0))
def _toggle_tab(self, command: str, tab_id: int | None) -> int:
"""Run a tab toggle command (mute/pin/...) and return the target tab ID."""
result = self._cmd(command, {"tabId": tab_id})
return self._field(result, "tabId", tab_id, fallback=int(tab_id or 0))
def multi_count(self, command: str, args: dict | None = None) -> "int | BrowserCounts":
"""Count command that aggregates into :class:`BrowserCounts` in multi-browser mode."""
multi_results = self._collect_multi_browser(command, args or {})
if not multi_results:
return self._client.dispatch(command, args or {})
by_browser = {target.display_name: int(count or 0) for target, count in multi_results}
return BrowserCounts(total=sum(by_browser.values()), by_browser=by_browser)
def _multi_count(self, command: str, args: dict | None = None) -> "int | BrowserCounts":
"""Count command that aggregates into :class:`BrowserCounts` in multi-browser mode."""
multi_results = self._collect_multi_browser(command, args or {})
if not multi_results:
return self._cmd(command, args or {})
by_browser = {target.display_name: int(count or 0) for target, count in multi_results}
return BrowserCounts(total=sum(by_browser.values()), by_browser=by_browser)
def multi_list(self, command: str, args: dict | None, mapper):
"""List command, flattening per-browser results in multi-browser mode.
def _multi_list(self, command: str, args: dict | None, mapper):
"""List command, flattening per-browser results in multi-browser mode.
*mapper* is ``(item, target) -> mapped`` where ``target`` is the source
:class:`BrowserTarget` in multi mode, or ``None`` in single-browser mode.
"""
multi_results = self._collect_multi_browser(command, args or {})
if multi_results:
return [
mapper(item, target)
for target, items in multi_results
for item in (items or [])
]
return [mapper(item, None) for item in (self._client.dispatch(command, args or {}) or [])]
*mapper* is ``(item, target) -> mapped`` where ``target`` is the source
:class:`BrowserTarget` in multi mode, or ``None`` in single-browser mode.
"""
multi_results = self._collect_multi_browser(command, args or {})
if multi_results:
return [
mapper(item, target)
for target, items in multi_results
for item in (items or [])
]
return [mapper(item, None) for item in (self._cmd(command, args or {}) or [])]
def apply_tab_filter(self, filter_fn: Callable[[Tab], bool] | Callable[[list[Tab]], Iterable[Tab]]) -> list[Tab]:
tabs = self._client.tabs.list()
def _apply_tab_filter(self, filter_fn: Callable[[Tab], bool] | Callable[[list[Tab]], Iterable[Tab]]) -> list[Tab]:
tabs = self.tabs.list()
try:
transformed = filter_fn(tabs)
except (AttributeError, TypeError):
return [tab for tab in tabs if filter_fn(tab)]
try:
transformed = filter_fn(tabs)
except (AttributeError, TypeError):
return [tab for tab in tabs if filter_fn(tab)]
if isinstance(transformed, list):
return transformed
if isinstance(transformed, tuple):
return list(transformed)
if isinstance(transformed, set):
return list(transformed)
if transformed is tabs:
return tabs
if isinstance(transformed, bool):
return [tab for tab in tabs if filter_fn(tab)]
if isinstance(transformed, list):
return transformed
if isinstance(transformed, tuple):
return list(transformed)
if isinstance(transformed, set):
return list(transformed)
if transformed is tabs:
return tabs
if isinstance(transformed, bool):
return [tab for tab in tabs if filter_fn(tab)]
try:
return list(transformed)
except TypeError:
return [tab for tab in tabs if filter_fn(tab)]
try:
return list(transformed)
except TypeError:
return [tab for tab in tabs if filter_fn(tab)]
browser_cli/sdk/session.py
+51 -50
View File
@@ -1,63 +1,64 @@
"""Session namespace: ``b.session.*``."""
from __future__ import annotations
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.base import Namespace, sdk_command
def _load_args(name, gentle_mode, discard_background_tabs, lazy, eager_tabs) -> dict:
return {
"name": name,
"gentleMode": gentle_mode,
"discardBackgroundTabs": discard_background_tabs,
"lazy": lazy,
"eagerTabs": eager_tabs,
}
class SessionNS(Namespace):
"""Save, restore, list, and diff browser sessions."""
"""Save, restore, list, and diff browser sessions."""
def save(self, name: str) -> dict:
"""Save all current tabs as session *name*. Returns the save result (incl. tab count)."""
return self._c._cmd("session.save", {"name": name}) or {}
@sdk_command("session.save", lambda self, name: {"name": name}, default={})
def save(self, name: str) -> dict:
"""Save all current tabs as session *name*. Returns the save result (incl. tab count)."""
@staticmethod
def _load_args(name, gentle_mode, discard_background_tabs, lazy, eager_tabs) -> dict:
return {
"name": name,
"gentleMode": gentle_mode,
"discardBackgroundTabs": discard_background_tabs,
"lazy": lazy,
"eagerTabs": eager_tabs,
}
def load(
self,
name: str,
*,
gentle_mode: str = "auto",
discard_background_tabs: bool = False,
lazy: bool = False,
eager_tabs: int = 10,
) -> dict:
"""Restore session *name*. Returns the load result (incl. tabs opened)."""
return self.command("session.load", _load_args(name, gentle_mode, discard_background_tabs, lazy, eager_tabs)) or {}
def load(
self,
name: str,
*,
gentle_mode: str = "auto",
discard_background_tabs: bool = False,
lazy: bool = False,
eager_tabs: int = 10,
) -> dict:
"""Restore session *name*. Returns the load result (incl. tabs opened)."""
args = self._load_args(name, gentle_mode, discard_background_tabs, lazy, eager_tabs)
return self._c._cmd("session.load", args) or {}
def load_background(
self,
name: str,
*,
gentle_mode: str = "auto",
discard_background_tabs: bool = False,
lazy: bool = False,
eager_tabs: int = 10,
) -> dict:
"""Restore session *name* as a background job. Returns the job descriptor."""
args = _load_args(name, gentle_mode, discard_background_tabs, lazy, eager_tabs)
return self.command("session.load", {**args, "__background": True}) or {}
def load_background(
self,
name: str,
*,
gentle_mode: str = "auto",
discard_background_tabs: bool = False,
lazy: bool = False,
eager_tabs: int = 10,
) -> dict:
"""Restore session *name* as a background job. Returns the job descriptor."""
args = self._load_args(name, gentle_mode, discard_background_tabs, lazy, eager_tabs)
return self._c._cmd("session.load", {**args, "__background": True}) or {}
@sdk_command("session.diff", lambda self, name_a, name_b: {"nameA": name_a, "nameB": name_b}, default={})
def diff(self, name_a: str, name_b: str) -> dict:
"""Diff two saved sessions."""
def diff(self, name_a: str, name_b: str) -> dict:
return self._c._cmd("session.diff", {"nameA": name_a, "nameB": name_b}) or {}
def list(self) -> list[dict]:
"""Return saved sessions.
def list(self) -> list[dict]:
"""Return saved sessions.
In implicit multi-browser mode each session dict includes a ``browser`` key.
"""
return self.multi_list("session.list", {}, self.tag_browser)
In implicit multi-browser mode each session dict includes a ``browser`` key.
"""
return self._c._multi_list("session.list", {}, self._c._tag_browser)
@sdk_command("session.remove", lambda self, name: {"name": name}, return_result=False)
def remove(self, name: str) -> None:
"""Remove a saved session."""
def remove(self, name: str) -> None:
self._c._cmd("session.remove", {"name": name})
def auto_save(self, enabled: bool) -> None:
self._c._cmd("session.auto_save", {"enabled": enabled})
@sdk_command("session.auto_save", lambda self, enabled: {"enabled": enabled}, return_result=False)
def auto_save(self, enabled: bool) -> None:
"""Enable or disable automatic session saves."""
browser_cli/sdk/tabs.py
+171 -175
View File
@@ -2,214 +2,210 @@
from __future__ import annotations
from collections.abc import Callable, Iterable
from typing import TYPE_CHECKING
from browser_cli.models import Tab
from browser_cli.models import BrowserCounts, Tab
from browser_cli.sdk.base import Namespace
if TYPE_CHECKING:
from browser_cli import BrowserCounts
class TabsNS(Namespace):
"""List, open, close, move, and inspect browser tabs."""
"""List, open, close, move, and inspect browser tabs."""
def list(self) -> list[Tab]:
"""Return all open tabs across all windows.
def list(self) -> list[Tab]:
"""Return all open tabs across all windows.
When multiple browsers are active and no browser was specified, each Tab
includes ``tab.browser`` naming its source browser.
"""
return self._c._multi_list("tabs.list", {}, self._c._make_tab_for)
When multiple browsers are active and no browser was specified, each Tab
includes ``tab.browser`` naming its source browser.
"""
return self.multi_list("tabs.list", {}, self.tab_from_target)
def open(
self,
url: str,
*,
wait: bool = False,
timeout: float = 30.0,
background: bool = False,
window: str | None = None,
group: str | None = None,
) -> Tab:
"""Open *url* in a new tab and return a bound :class:`Tab`.
def open(
self,
url: str,
*,
wait: bool = False,
timeout: float = 30.0,
background: bool = False,
window: str | None = None,
group: str | None = None,
) -> Tab:
"""Open *url* in a new tab and return a bound :class:`Tab`.
Set ``wait=True`` to block until the page reaches ``readyState=complete``.
"""
if wait:
return self._c.nav.open_wait(url, timeout=timeout, background=background, window=window, group=group)
return self._c._require_tab(
self._c._cmd("navigate.open", {"url": url, "background": background, "window": window, "group": group}),
"navigate.open returned unexpected data",
)
Set ``wait=True`` to block until the page reaches ``readyState=complete``.
"""
if wait:
return self._c.nav.open_wait(url, timeout=timeout, background=background, window=window, group=group)
return self.require_tab(
self.command("navigate.open", {"url": url, "background": background, "window": window, "group": group}),
"navigate.open returned unexpected data",
)
def get(self, tab_id: int) -> Tab:
"""Return a specific tab by ID."""
return self.status(tab_id)
def get(self, tab_id: int) -> Tab:
"""Return a specific tab by ID."""
return self.status(tab_id)
def active(self) -> Tab:
"""Return the active tab."""
return self.status()
def active(self) -> Tab:
"""Return the active tab."""
return self.status()
def query(self, search: str) -> list[Tab]:
"""Search tabs by URL or title."""
return [self._c._make_tab(t) for t in (self._c._cmd("tabs.query", {"search": search}) or [])]
def query(self, search: str) -> list[Tab]:
"""Search tabs by URL or title."""
return [self.tab_from(t) for t in (self.command("tabs.query", {"search": search}) or [])]
def first(self, search: str) -> Tab | None:
"""Return the first tab matching *search*, or ``None``."""
matches = self.query(search)
return matches[0] if matches else None
def first(self, search: str) -> Tab | None:
"""Return the first tab matching *search*, or ``None``."""
matches = self.query(search)
return matches[0] if matches else None
def close(
self,
tab_id: int | None = None,
*,
tab_ids: Iterable[int | Tab] | None = None,
inactive: bool = False,
duplicates: bool = False,
gentle_mode: str = "auto",
) -> int:
"""Close tab(s). Returns the number of tabs closed.
def close(
self,
tab_id: int | None = None,
*,
tab_ids: Iterable[int | Tab] | None = None,
inactive: bool = False,
duplicates: bool = False,
gentle_mode: str = "auto",
) -> int:
"""Close tab(s). Returns the number of tabs closed.
Pass ``tab_ids`` to close many tabs in a single round-trip. Accepts tab
IDs or :class:`Tab` objects. ``gentle_mode`` (auto/normal/gentle/ultra)
controls throttling of large close operations.
"""
ids = None
if tab_ids is not None:
ids = [t.id if isinstance(t, Tab) else t for t in tab_ids]
result = self._c._cmd("tabs.close", {
"tabId": tab_id,
"tabIds": ids,
"inactive": inactive,
"duplicates": duplicates,
"gentleMode": gentle_mode,
})
return self._c._field(result, "closed", 1)
Pass ``tab_ids`` to close many tabs in a single round-trip. Accepts tab
IDs or :class:`Tab` objects. ``gentle_mode`` (auto/normal/gentle/ultra)
controls throttling of large close operations.
"""
ids = None
if tab_ids is not None:
ids = [t.id if isinstance(t, Tab) else t for t in tab_ids]
result = self.command("tabs.close", {
"tabId": tab_id,
"tabIds": ids,
"inactive": inactive,
"duplicates": duplicates,
"gentleMode": gentle_mode,
})
return self.field(result, "closed", 1)
def close_inactive(self) -> int:
"""Close all inactive tabs. Returns count closed."""
return self._c._field(self._c._cmd("tabs.close", {"inactive": True}), "closed", 0)
def close_inactive(self) -> int:
"""Close all inactive tabs. Returns count closed."""
return self.field(self.command("tabs.close", {"inactive": True}), "closed", 0)
def close_duplicates(self) -> int:
"""Close duplicate tabs. Returns count closed."""
return self._c._field(self._c._cmd("tabs.close", {"duplicates": True}), "closed", 0)
def close_duplicates(self) -> int:
"""Close duplicate tabs. Returns count closed."""
return self.field(self.command("tabs.close", {"duplicates": True}), "closed", 0)
def move(
self, tab_id: int, *,
forward: bool = False, backward: bool = False,
group_id: int | None = None, window_id: int | None = None, index: int | None = None,
) -> None:
self._c._cmd("tabs.move", {
"tabId": tab_id, "forward": forward, "backward": backward,
"groupId": group_id, "windowId": window_id, "index": index,
})
def move(
self, tab_id: int, *,
forward: bool = False, backward: bool = False,
group_id: int | None = None, window_id: int | None = None, index: int | None = None,
) -> None:
self.command("tabs.move", {
"tabId": tab_id, "forward": forward, "backward": backward,
"groupId": group_id, "windowId": window_id, "index": index,
})
def activate(self, tab_id: int) -> None:
"""Switch browser focus to a tab by ID."""
self._c._cmd("tabs.active", {"tabId": tab_id})
def activate(self, tab_id: int) -> None:
"""Switch browser focus to a tab by ID."""
self.command("tabs.active", {"tabId": tab_id})
def status(self, tab_id: int | None = None) -> Tab:
"""Return status for the active tab or a specific tab."""
return self._c._require_tab(self._c._cmd("tabs.status", {"tabId": tab_id}), "No tab status returned")
def status(self, tab_id: int | None = None) -> Tab:
"""Return status for the active tab or a specific tab."""
return self.require_tab(self.command("tabs.status", {"tabId": tab_id}), "No tab status returned")
def mute(self, tab_id: int | None = None) -> int:
"""Mute the active tab or a specific tab. Returns the target tab ID."""
return self._c._toggle_tab("tabs.mute", tab_id)
def mute(self, tab_id: int | None = None) -> int:
"""Mute the active tab or a specific tab. Returns the target tab ID."""
return self.toggle_tab("tabs.mute", tab_id)
def unmute(self, tab_id: int | None = None) -> int:
"""Unmute the active tab or a specific tab. Returns the target tab ID."""
return self._c._toggle_tab("tabs.unmute", tab_id)
def unmute(self, tab_id: int | None = None) -> int:
"""Unmute the active tab or a specific tab. Returns the target tab ID."""
return self.toggle_tab("tabs.unmute", tab_id)
def pin(self, tab_id: int | None = None) -> int:
"""Pin the active tab or a specific tab. Returns the target tab ID."""
return self._c._toggle_tab("tabs.pin", tab_id)
def pin(self, tab_id: int | None = None) -> int:
"""Pin the active tab or a specific tab. Returns the target tab ID."""
return self.toggle_tab("tabs.pin", tab_id)
def unpin(self, tab_id: int | None = None) -> int:
"""Unpin the active tab or a specific tab. Returns the target tab ID."""
return self._c._toggle_tab("tabs.unpin", tab_id)
def unpin(self, tab_id: int | None = None) -> int:
"""Unpin the active tab or a specific tab. Returns the target tab ID."""
return self.toggle_tab("tabs.unpin", tab_id)
def watch_url(self, pattern: str, *, tab_id: int | None = None, timeout: float = 30.0) -> Tab:
"""Block until the tab URL matches regex pattern. Returns the Tab."""
return self._c._require_tab(
self._c._cmd("tabs.watch_url", {"pattern": pattern, "tabId": tab_id, "timeout": int(timeout * 1000)}),
"tabs.watch_url returned unexpected data",
)
def watch_url(self, pattern: str, *, tab_id: int | None = None, timeout: float = 30.0) -> Tab:
"""Block until the tab URL matches regex pattern. Returns the Tab."""
return self.require_tab(
self.command("tabs.watch_url", {"pattern": pattern, "tabId": tab_id, "timeout": int(timeout * 1000)}),
"tabs.watch_url returned unexpected data",
)
def wait_for_load(
self,
tab_id: int | None = None,
*,
timeout: float = 30.0,
ready_state: str = "complete",
) -> Tab:
"""Block until the tab finishes loading. Returns the Tab when ready.
def wait_for_load(
self,
tab_id: int | None = None,
*,
timeout: float = 30.0,
ready_state: str = "complete",
) -> Tab:
"""Block until the tab finishes loading. Returns the Tab when ready.
Args:
tab_id: Tab to watch. Defaults to the active tab.
timeout: Max seconds to wait before raising ``RuntimeError``.
ready_state: ``"complete"`` (default) or ``"interactive"``.
"""
return self._c._require_tab(
self._c._cmd("navigate.wait", {
"tabId": tab_id,
"timeout": int(timeout * 1000),
"readyState": ready_state,
}),
"navigate.wait returned unexpected data",
)
Args:
tab_id: Tab to watch. Defaults to the active tab.
timeout: Max seconds to wait before raising ``RuntimeError``.
ready_state: ``"complete"`` (default) or ``"interactive"``.
"""
return self.require_tab(
self.command("navigate.wait", {
"tabId": tab_id,
"timeout": int(timeout * 1000),
"readyState": ready_state,
}),
"navigate.wait returned unexpected data",
)
def screenshot(
self,
tab_id: int | None = None,
*,
format: str = "png",
quality: int | None = None,
) -> str:
"""Capture the visible area of a tab. Returns a base64 data URL.
def screenshot(
self,
tab_id: int | None = None,
*,
format: str = "png",
quality: int | None = None,
) -> str:
"""Capture the visible area of a tab. Returns a base64 data URL.
Args:
tab_id: Tab to capture. Defaults to the active tab.
format: ``"png"`` (default) or ``"jpeg"``.
quality: JPEG quality 0-100 (ignored for PNG).
"""
result = self._c._cmd("tabs.screenshot", {"tabId": tab_id, "format": format, "quality": quality})
return self._c._field(result, "dataUrl", "", fallback=str(result))
Args:
tab_id: Tab to capture. Defaults to the active tab.
format: ``"png"`` (default) or ``"jpeg"``.
quality: JPEG quality 0-100 (ignored for PNG).
"""
result = self.command("tabs.screenshot", {"tabId": tab_id, "format": format, "quality": quality})
return self.field(result, "dataUrl", "", fallback=str(result))
def active_in_window(self, window_id: int) -> Tab:
"""Return active tab for a specific browser window."""
return self._c._require_tab(
self._c._cmd("tabs.active_in_window", {"windowId": window_id}),
f"No active tab found for window {window_id}",
)
def active_in_window(self, window_id: int) -> Tab:
"""Return active tab for a specific browser window."""
return self.require_tab(
self.command("tabs.active_in_window", {"windowId": window_id}),
f"No active tab found for window {window_id}",
)
def filter(
self,
pattern_or_filter: str | Callable[[Tab], bool] | Callable[[list[Tab]], Iterable[Tab]],
) -> list[Tab]:
"""Return tabs filtered by URL pattern or a Python callable."""
if isinstance(pattern_or_filter, str):
return [self._c._make_tab(t) for t in (self._c._cmd("tabs.filter", {"pattern": pattern_or_filter}) or [])]
return self._c._apply_tab_filter(pattern_or_filter)
def filter(
self,
pattern_or_filter: str | Callable[[Tab], bool] | Callable[[list[Tab]], Iterable[Tab]],
) -> list[Tab]:
"""Return tabs filtered by URL pattern or a Python callable."""
if isinstance(pattern_or_filter, str):
return [self.tab_from(t) for t in (self.command("tabs.filter", {"pattern": pattern_or_filter}) or [])]
return self.apply_tab_filter(pattern_or_filter)
def count(self, pattern: str | None = None) -> "int | BrowserCounts":
"""Count open tabs, optionally filtered by URL pattern.
def count(self, pattern: str | None = None) -> "int | BrowserCounts":
"""Count open tabs, optionally filtered by URL pattern.
Returns ``BrowserCounts`` in implicit multi-browser mode.
"""
return self._c._multi_count("tabs.count", {"pattern": pattern})
Returns ``BrowserCounts`` in implicit multi-browser mode.
"""
return self.multi_count("tabs.count", {"pattern": pattern})
def html(self, tab_id: int | None = None) -> str:
"""Return the full HTML source of the active (or specified) tab."""
return self._c._cmd("tabs.html", {"tabId": tab_id}) or ""
def html(self, tab_id: int | None = None) -> str:
"""Return the full HTML source of the active (or specified) tab."""
return self.command("tabs.html", {"tabId": tab_id}) or ""
def dedupe(self, *, gentle_mode: str = "auto") -> int:
"""Close duplicate tabs (keep the first occurrence of each URL). Returns count closed."""
return self._c._field(self._c._cmd("tabs.dedupe", {"gentleMode": gentle_mode}), "closed", 0)
def dedupe(self, *, gentle_mode: str = "auto") -> int:
"""Close duplicate tabs (keep the first occurrence of each URL). Returns count closed."""
return self.field(self.command("tabs.dedupe", {"gentleMode": gentle_mode}), "closed", 0)
def sort(self, by: str = "domain", *, gentle_mode: str = "auto") -> None:
"""Sort tabs within each window. *by* is one of 'domain', 'title', 'time'."""
self._c._cmd("tabs.sort", {"by": by, "gentleMode": gentle_mode})
def sort(self, by: str = "domain", *, gentle_mode: str = "auto") -> None:
"""Sort tabs within each window. *by* is one of 'domain', 'title', 'time'."""
self.command("tabs.sort", {"by": by, "gentleMode": gentle_mode})
def merge_windows(self, *, gentle_mode: str = "auto") -> int:
"""Move all tabs into the focused window. Returns count moved."""
return self._c._field(self._c._cmd("tabs.merge_windows", {"gentleMode": gentle_mode}), "moved", 0)
def merge_windows(self, *, gentle_mode: str = "auto") -> int:
"""Move all tabs into the focused window. Returns count moved."""
return self.field(self.command("tabs.merge_windows", {"gentleMode": gentle_mode}), "moved", 0)
browser_cli/sdk/windows.py
+16 -14
View File
@@ -1,24 +1,26 @@
"""Windows namespace: ``b.windows.*``."""
from __future__ import annotations
from browser_cli.sdk.base import Namespace
from browser_cli.sdk.base import Namespace, sdk_command
class WindowsNS(Namespace):
"""List, open, close, and rename browser windows."""
"""List, open, close, and rename browser windows."""
def list(self) -> list[dict]:
"""Return browser windows.
def list(self) -> list[dict]:
"""Return browser windows.
In implicit multi-browser mode each window dict includes a ``browser`` key.
"""
return self._c._multi_list("windows.list", {}, self._c._tag_browser)
In implicit multi-browser mode each window dict includes a ``browser`` key.
"""
return self.multi_list("windows.list", {}, self.tag_browser)
def open(self, url: str | None = None) -> dict:
"""Open a new browser window, optionally on a URL."""
return self._c._cmd("windows.open", {"url": url}) or {}
@sdk_command("windows.open", lambda self, url=None: {"url": url}, default={})
def open(self, url: str | None = None) -> dict:
"""Open a new browser window, optionally on a URL."""
def close(self, window_id: int) -> None:
self._c._cmd("windows.close", {"windowId": window_id})
@sdk_command("windows.close", lambda self, window_id: {"windowId": window_id}, return_result=False)
def close(self, window_id: int) -> None:
"""Close a browser window by ID."""
def rename(self, window_id: int, name: str) -> None:
self._c._cmd("windows.rename", {"windowId": window_id, "name": name})
@sdk_command("windows.rename", lambda self, window_id, name: {"windowId": window_id, "name": name}, return_result=False)
def rename(self, window_id: int, name: str) -> None:
"""Rename a browser window."""
browser_cli/sdk/workflow_decorators.py
+198
View File
@@ -0,0 +1,198 @@
"""Shared workflow decorator implementation for sync and async SDK clients."""
from __future__ import annotations
import functools
import time
from collections.abc import Callable
from typing import TypeVar
F = TypeVar("F", bound=Callable)
_NO_INJECT = object()
class WorkflowDecoratorsMixin:
"""Shared implementation for sync and async workflow decorators.
Subclasses only define *how* browser calls, user functions, and sleeps are
executed. The actual decorators live once here, so sync and async SDKs stay
in lockstep.
"""
_c: object
@staticmethod
def _inject(kwargs: dict, keyword: str | None, value):
if keyword is not None:
kwargs[keyword] = value
return (), kwargs
return (value,), kwargs
def _run(self, func: Callable, *args, **kwargs):
return func(*args, **kwargs)
def _call_wrapped(self, func: Callable, *args, **kwargs):
return func(*args, **kwargs)
def _sleep(self, delay: float) -> None:
time.sleep(delay)
def _value_decorator(
self,
func: F | None,
get_value: Callable,
*,
keyword: str | None | object = "tab",
cleanup: Callable | None = None,
):
"""Build a decorator around a browser-side value lookup.
``get_value`` always runs before the wrapped function. If ``keyword`` is
``_NO_INJECT`` the value is only used by ``cleanup`` and is not passed to
the wrapped function. ``keyword=None`` injects it positionally.
"""
def decorator(fn: F) -> F:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
value = self._run(get_value)
try:
extra_args = ()
if keyword is not _NO_INJECT:
extra_args, kwargs = self._inject(kwargs, keyword, value)
return self._call_wrapped(fn, *extra_args, *args, **kwargs)
finally:
if cleanup is not None:
self._run(cleanup, value)
return wrapper # type: ignore[return-value]
return decorator(func) if func is not None else decorator
def active_tab(self, func: F | None = None, *, keyword: str | None = "tab"):
"""Decorate a function so it receives the current active tab.
By default the tab is injected as ``tab=...``. Pass ``keyword=None`` to
pass it as the first positional argument instead.
"""
return self._value_decorator(func, self._c.tabs.active, keyword=keyword) # type: ignore[attr-defined]
def new_tab(
self,
url: str,
*,
wait: bool = False,
timeout: float = 30.0,
background: bool = False,
window: str | None = None,
group: str | None = None,
close: bool = False,
keyword: str | None = "tab",
):
"""Open *url* for the wrapped function and inject the created tab.
Set ``close=True`` to close the tab in a ``finally`` block after the
wrapped function returns or raises.
"""
def open_tab():
return self._c.tabs.open( # type: ignore[attr-defined]
url,
wait=wait,
timeout=timeout,
background=background,
window=window,
group=group,
)
def close_tab(tab):
tab.close()
return self._value_decorator(None, open_tab, keyword=keyword, cleanup=close_tab if close else None)
def wait_for_selector(
self,
selector: str,
*,
timeout: float = 10.0,
visible: bool = False,
hidden: bool = False,
tab_id: int | None = None,
keyword: str | None = None,
):
"""Wait for a selector before calling the wrapped function.
Pass ``keyword="result"`` (or similar) to inject the wait result into
the wrapped function. By default the result is not injected.
"""
def wait():
return self._c.dom.wait_for( # type: ignore[attr-defined]
selector,
timeout=timeout,
visible=visible,
hidden=hidden,
tab_id=tab_id,
)
inject = keyword if keyword is not None else _NO_INJECT
return self._value_decorator(None, wait, keyword=inject)
def wait_for_url(
self,
pattern: str,
*,
tab_id: int | None = None,
timeout: float = 30.0,
keyword: str | None = "tab",
):
"""Wait until a tab URL matches *pattern* before calling the function."""
def wait():
return self._c.tabs.watch_url(pattern, tab_id=tab_id, timeout=timeout) # type: ignore[attr-defined]
inject = keyword if keyword is not None else _NO_INJECT
return self._value_decorator(None, wait, keyword=inject)
def performance_profile(self, profile: str, *, restore: bool = True):
"""Temporarily set the extension performance profile around a function."""
def decorator(fn: F) -> F:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
previous = None
if restore:
previous = self._run(self._c.perf.status).get("performanceProfile") # type: ignore[attr-defined]
self._run(self._c.perf.set_profile, profile) # type: ignore[attr-defined]
try:
return self._call_wrapped(fn, *args, **kwargs)
finally:
if previous:
self._run(self._c.perf.set_profile, previous) # type: ignore[attr-defined]
return wrapper # type: ignore[return-value]
return decorator
def save_session_before(self, name: str):
"""Save the current browser session before running the function."""
return self._value_decorator(None, lambda: self._c.session.save(name), keyword=_NO_INJECT) # type: ignore[attr-defined]
def retry(
self,
*,
times: int = 3,
delay: float = 0.0,
exceptions: tuple[type[BaseException], ...] = (Exception,),
):
"""Retry the wrapped function when it raises one of *exceptions*."""
attempts = max(1, times)
def decorator(fn: F) -> F:
@functools.wraps(fn)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(attempts):
try:
return self._call_wrapped(fn, *args, **kwargs)
except exceptions as exc:
last_error = exc
if attempt == attempts - 1:
raise
if delay > 0:
self._sleep(delay)
raise last_error # type: ignore[misc]
return wrapper # type: ignore[return-value]
return decorator