daniel156161
7cb2a8b618
Testing / remote-protocol-compat (0.9.5) (push) Successful in 1m4s
Testing / test (push) Successful in 1m22s
Testing / remote-protocol-compat (0.9.3) (push) Successful in 1m7s
Package Extension / package-extension (push) Successful in 1m1s
Build & Publish Package / publish (push) Successful in 1m5s
- Split auth into focused package modules for agent keys, file keys, signing, and post-quantum transport helpers while keeping the public browser_cli.auth import surface intact. - Move transport encoding internals into a package with separate codec and binary-hoisting helpers, preserving browser_cli.transport compatibility. - Extract remote TCP auth/socket helpers and serve challenge setup out of the runtime paths to make connection handling easier to reason about. - Move the extension markdown extractor into a dedicated content/markdown folder with separate root selection, code normalization, renderer, and utils. - Centralize CLI Rich rendering helpers for tab/window tree and table output, and add rendering tests for the shared builders. - Remove local typing ignores in SDK/decorator/script plumbing and bump the package and extension version to 0.15.3.
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
"""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, cast
|
|
|
|
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 cast(F, async_wrapper)
|
|
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 cast(F, async_wrapper)
|
|
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 cast(BaseException, last_error)
|
|
return cast(F, async_wrapper)
|
|
return WorkflowDecoratorsMixin.retry(self, times=times, delay=delay, exceptions=exceptions)(fn)
|
|
return decorator
|