tempestweb.native

The browser's Web APIs as typed Python awaitables — geolocation, clipboard, camera, HTTP, storage, sensors, WebPush. The same await native.<capability>() call works in Modes A and B; only who runs it on the far side of the bridge changes.

Guide with examples: Native capabilities · Native capability reference · Native event channel.

tempestweb.native

tempestweb.native — typed Python wrappers over browser Web APIs (Track N).

The web sibling of tempestroid.native: each device/web capability is a typed Python awaitable that application code calls without caring whether its Python runs in the browser (Mode A / WASM) or on the server (Mode B). The single seam that differs between the modes is the installed :class:NativeBridge — see :mod:tempestweb.native.dispatch for the full Mode-A vs Mode-B explanation and client/native/*.js for the browser glue.

Capabilities are exposed two ways. Import the module for the plan-facing namespaced calls::

from tempestweb import native

res = await native.http.request("GET", "/api/items")
pos = await native.geolocation.get()
await native.audio.play("/audio/plim.wav", volume=0.4)
result = await native.share(title="Hi", url="https://example.com")
photo = await native.camera.capture()

or import the symbols directly::

from tempestweb.native import request, get_position, ShareResult

Capabilities:

• http (N0) — :func:~tempestweb.native.http.request (retry + backoff + idempotency), :func:~tempestweb.native.http.upload, :func:~tempestweb.native.http.poll, :func:~tempestweb.native.http.generate_idempotency_key.
• audio (N1) — :func:~tempestweb.native.audio.play / stop.
• share (N2) — :func:~tempestweb.native.share.share / :func:~tempestweb.native.share.is_share_supported.
• geolocation / clipboard / storage (N3) — :func:~tempestweb.native.geolocation.get, clipboard.read/write, storage.put/get/list_keys/remove (layered over IndexedDB).
• camera (N4) — :func:~tempestweb.native.camera.capture.
• notifications — :func:~tempestweb.native.notifications.notify / request_permission.

PlayResult

Bases: BaseModel

The outcome of a :func:play call.

Attributes:

Name	Type	Description
played	bool	Whether playback actually started.
blocked	bool	Whether the browser blocked autoplay (no user gesture yet). When True, played is False and no error is raised.
channel	str	The channel the sound was routed to.

Source code in tempestweb/native/audio.py

class PlayResult(BaseModel):
    """The outcome of a :func:`play` call.

    Attributes:
        played: Whether playback actually started.
        blocked: Whether the browser blocked autoplay (no user gesture yet). When
            ``True``, ``played`` is ``False`` and no error is raised.
        channel: The channel the sound was routed to.
    """

    model_config = ConfigDict(frozen=True)

    played: bool
    blocked: bool = False
    channel: str = "default"

BatteryStatus

Bases: BaseModel

A snapshot of the device battery reported by the Battery Status API.

Attributes:

Name	Type	Description
level	float	The charge level as a fraction from 0.0 (empty) to 1.0 (full).
charging	bool	Whether the battery is currently charging.
charging_time	float	Seconds until the battery is fully charged (0.0 when already full, inf when unknown).
discharging_time	float	Seconds until the battery is empty (inf when unknown or while charging).

Source code in tempestweb/native/battery.py

class BatteryStatus(BaseModel):
    """A snapshot of the device battery reported by the Battery Status API.

    Attributes:
        level: The charge level as a fraction from ``0.0`` (empty) to ``1.0`` (full).
        charging: Whether the battery is currently charging.
        charging_time: Seconds until the battery is fully charged (``0.0`` when
            already full, ``inf`` when unknown).
        discharging_time: Seconds until the battery is empty (``inf`` when unknown
            or while charging).
    """

    model_config = ConfigDict(frozen=True)

    level: float
    charging: bool
    charging_time: float
    discharging_time: float

BluetoothDevice dataclass

A Bluetooth device paired through the Web Bluetooth API.

Attributes:

Name	Type	Description
id	str	The opaque device id; the client holds the live BluetoothDevice in a registry keyed by this id.
name	str	The device's advertised name, or "" when unknown.

Source code in tempestweb/native/bluetooth.py

@dataclass(frozen=True)
class BluetoothDevice:
    """A Bluetooth device paired through the Web Bluetooth API.

    Attributes:
        id: The opaque device id; the client holds the live ``BluetoothDevice`` in a
            registry keyed by this id.
        name: The device's advertised name, or ``""`` when unknown.
    """

    id: str
    name: str

FFIBridge

Mode A bridge: call client/native/*.js in-process via Pyodide FFI.

Under Pyodide, client/native/index.js exposes a single async dispatch function on the page (window.__tempestweb_native__(envelope)) returning a JS promise that resolves to a native_result envelope. This bridge awaits that promise directly — Python and the Web API share the browser's one event loop, so there is no serialization and no round-trip.

The JS callable is injected (rather than reached through a hard import js) so the dispatch logic is unit-testable with a fake async callable that mimics the FFI contract.

Attributes:

Name	Type	Description
dispatch	Callable[[str], Awaitable[str]]	The injected async JS callable (envelope) -> native_result.

Source code in tempestweb/native/bridges.py

class FFIBridge:
    """Mode A bridge: call ``client/native/*.js`` in-process via Pyodide FFI.

    Under Pyodide, ``client/native/index.js`` exposes a single async dispatch
    function on the page (``window.__tempestweb_native__(envelope)``) returning a
    JS promise that resolves to a ``native_result`` envelope. This bridge awaits
    that promise directly — Python and the Web API share the browser's one event
    loop, so there is no serialization and no round-trip.

    The JS callable is injected (rather than reached through a hard ``import js``)
    so the dispatch logic is unit-testable with a fake async callable that mimics
    the FFI contract.

    Attributes:
        dispatch: The injected async JS callable ``(envelope) -> native_result``.
    """

    def __init__(
        self,
        dispatch: Callable[[str], Awaitable[str]],
        subscribe_js: (
            Callable[[str, Callable[[str], None]], Awaitable[str]] | None
        ) = None,
        unsubscribe_js: Callable[[str], Awaitable[None]] | None = None,
    ) -> None:
        """Initialize the FFI bridge.

        Args:
            dispatch: An awaitable callable that takes a ``native_call`` envelope as
                a **JSON string** and resolves to the ``native_result`` envelope as
                a JSON string. Strings are used (not dicts) because they cross the
                Pyodide FFI boundary cleanly — no PyProxy/JsProxy conversion, the
                same convention as the patch/event callbacks. In a real browser this
                wraps ``window.__tempestweb_native__``; in tests it is a fake.
            subscribe_js: Awaitable callable ``(envelope_json, emit) -> sub_id`` that
                opens an event-channel subscription (T-EV), wrapping
                ``window.__tempestweb_native_subscribe__``. ``emit`` is a Python
                callback the browser invokes with each event as a **JSON string**.
                ``None`` when the Mode-A bootstrap has not wired streaming.
            unsubscribe_js: Awaitable callable ``(sub_id) -> None`` closing a
                subscription, wrapping ``window.__tempestweb_native_unsubscribe__``.
        """
        self.dispatch: Callable[[str], Awaitable[str]] = dispatch
        self.subscribe_js: (
            Callable[[str, Callable[[str], None]], Awaitable[str]] | None
        ) = subscribe_js
        self.unsubscribe_js: Callable[[str], Awaitable[None]] | None = unsubscribe_js

    async def call(self, envelope: dict[str, Any]) -> dict[str, Any]:
        """Dispatch a ``native_call`` envelope and await the JS promise result.

        The envelope crosses to JS as a JSON string and the ``native_result`` comes
        back as one, so nothing relies on FFI object conversion.

        Args:
            envelope: A ``native_call`` envelope carrying a ``call_id``.

        Returns:
            The ``native_result`` envelope ``client/native/*.js`` resolved with.
        """
        raw: str = await self.dispatch(json.dumps(envelope))
        result: dict[str, Any] = json.loads(raw)
        return result

    async def subscribe(
        self,
        capability: str,
        args: dict[str, Any],
        emit: Callable[[dict[str, Any]], None],
    ) -> str:
        """Open an event-channel subscription in-process via the JS FFI (T-EV).

        The subscribe envelope crosses to JS as a JSON string; the browser calls
        the wrapped ``emit`` with each event as a JSON string, which this method
        parses back into a ``dict`` before handing it to the Python ``emit``.

        Args:
            capability: The dotted streaming capability name.
            args: JSON-able subscription arguments.
            emit: Callback invoked with each ``{"event"|"error"|"done": ...}`` dict.

        Returns:
            The subscription id.

        Raises:
            BrowserUnavailableError: If Mode-A streaming was not wired at bootstrap.
        """
        if self.subscribe_js is None:
            raise BrowserUnavailableError(
                "mode A native event channel is not wired (no subscribe callable)"
            )
        sub_id = _next_sub_id()
        envelope = native_subscribe(capability, args, sub_id)

        def emit_str(raw: str) -> None:
            """Decode one streamed event and pass it to the Python subscriber.

            The JS side sends a JSON **string** because strings cross
            ``pyodide.ffi`` without proxy conversion; this is where that wire
            form becomes the dict the caller subscribed for.

            Args:
                raw: The ``native_event`` payload, JSON-encoded.
            """
            emit(json.loads(raw))

        await self.subscribe_js(json.dumps(envelope), emit_str)
        return sub_id

    async def unsubscribe(self, sub_id: str) -> None:
        """Close an event-channel subscription via the JS FFI.

        Args:
            sub_id: The id returned by :meth:`subscribe`.
        """
        if self.unsubscribe_js is not None:
            await self.unsubscribe_js(sub_id)

call async

call(envelope: dict[str, Any]) -> dict[str, Any]

Dispatch a native_call envelope and await the JS promise result.

The envelope crosses to JS as a JSON string and the native_result comes back as one, so nothing relies on FFI object conversion.

Parameters:

Name	Type	Description	Default
envelope	dict[str, Any]	A native_call envelope carrying a call_id.	required

Returns:

Type	Description
dict[str, Any]	The native_result envelope client/native/*.js resolved with.

Source code in tempestweb/native/bridges.py

async def call(self, envelope: dict[str, Any]) -> dict[str, Any]:
    """Dispatch a ``native_call`` envelope and await the JS promise result.

    The envelope crosses to JS as a JSON string and the ``native_result`` comes
    back as one, so nothing relies on FFI object conversion.

    Args:
        envelope: A ``native_call`` envelope carrying a ``call_id``.

    Returns:
        The ``native_result`` envelope ``client/native/*.js`` resolved with.
    """
    raw: str = await self.dispatch(json.dumps(envelope))
    result: dict[str, Any] = json.loads(raw)
    return result

subscribe async

subscribe(capability: str, args: dict[str, Any], emit: Callable[[dict[str, Any]], None]) -> str

Open an event-channel subscription in-process via the JS FFI (T-EV).

The subscribe envelope crosses to JS as a JSON string; the browser calls the wrapped emit with each event as a JSON string, which this method parses back into a dict before handing it to the Python emit.

Parameters:

Name	Type	Description	Default
capability	str	The dotted streaming capability name.	required
args	dict[str, Any]	JSON-able subscription arguments.	required
emit	Callable[[dict[str, Any]], None]	Callback invoked with each {"event"|"error"|"done": ...} dict.	required

Returns:

Type	Description
str	The subscription id.

Raises:

Type	Description
BrowserUnavailableError	If Mode-A streaming was not wired at bootstrap.

Source code in tempestweb/native/bridges.py

async def subscribe(
    self,
    capability: str,
    args: dict[str, Any],
    emit: Callable[[dict[str, Any]], None],
) -> str:
    """Open an event-channel subscription in-process via the JS FFI (T-EV).

    The subscribe envelope crosses to JS as a JSON string; the browser calls
    the wrapped ``emit`` with each event as a JSON string, which this method
    parses back into a ``dict`` before handing it to the Python ``emit``.

    Args:
        capability: The dotted streaming capability name.
        args: JSON-able subscription arguments.
        emit: Callback invoked with each ``{"event"|"error"|"done": ...}`` dict.

    Returns:
        The subscription id.

    Raises:
        BrowserUnavailableError: If Mode-A streaming was not wired at bootstrap.
    """
    if self.subscribe_js is None:
        raise BrowserUnavailableError(
            "mode A native event channel is not wired (no subscribe callable)"
        )
    sub_id = _next_sub_id()
    envelope = native_subscribe(capability, args, sub_id)

    def emit_str(raw: str) -> None:
        """Decode one streamed event and pass it to the Python subscriber.

        The JS side sends a JSON **string** because strings cross
        ``pyodide.ffi`` without proxy conversion; this is where that wire
        form becomes the dict the caller subscribed for.

        Args:
            raw: The ``native_event`` payload, JSON-encoded.
        """
        emit(json.loads(raw))

    await self.subscribe_js(json.dumps(envelope), emit_str)
    return sub_id

unsubscribe async

unsubscribe(sub_id: str) -> None

Close an event-channel subscription via the JS FFI.

Parameters:

Name	Type	Description	Default
sub_id	str	The id returned by :meth:subscribe.	required

Source code in tempestweb/native/bridges.py

async def unsubscribe(self, sub_id: str) -> None:
    """Close an event-channel subscription via the JS FFI.

    Args:
        sub_id: The id returned by :meth:`subscribe`.
    """
    if self.unsubscribe_js is not None:
        await self.unsubscribe_js(sub_id)

ProxyBridge

Mode B bridge: proxy native calls to the browser over the WS/SSE transport.

The server has no Web APIs of its own, so every native_call is forwarded to the thin client, which runs client/native/*.js against the browser Web API and posts a native_result back. This bridge translates the :class:~tempestweb.native.dispatch.NativeBridge contract into "send a native_call frame, await the matching native_result frame".

Attributes:

Name	Type	Description
send_frame	Callable[[dict[str, Any]], None]	Injected callable that ships a JSON-able frame to the client (the server session wires this to the patch transport's send path).
timeout	float | None	Seconds to wait for a native_result before giving up.

Source code in tempestweb/native/bridges.py

class ProxyBridge:
    """Mode B bridge: proxy native calls to the browser over the WS/SSE transport.

    The server has no Web APIs of its own, so every ``native_call`` is forwarded to
    the thin client, which runs ``client/native/*.js`` against the browser Web API
    and posts a ``native_result`` back. This bridge translates the
    :class:`~tempestweb.native.dispatch.NativeBridge` contract into "send a
    ``native_call`` frame, await the matching ``native_result`` frame".

    Attributes:
        send_frame: Injected callable that ships a JSON-able frame to the client
            (the server session wires this to the patch transport's send path).
        timeout: Seconds to wait for a ``native_result`` before giving up.
    """

    def __init__(
        self,
        send_frame: Callable[[dict[str, Any]], None],
        *,
        timeout: float | None = DEFAULT_NATIVE_CALL_TIMEOUT,
    ) -> None:
        """Initialize the proxy bridge.

        Args:
            send_frame: Callable shipping a ``native_call`` frame to the client over
                the transport. The client later posts a ``native_result`` frame
                back, which :meth:`resolve` matches to the pending future.
            timeout: How long to wait for the matching ``native_result``. ``None``
                waits forever, which is what the bridge used to do unconditionally:
                a client that never answers (a closed tab, a capability that threw
                before replying) left the awaiting handler suspended until the
                session ended.
        """
        self.send_frame: Callable[[dict[str, Any]], None] = send_frame
        self.timeout: float | None = timeout
        self._pending: dict[str, asyncio.Future[dict[str, Any]]] = {}
        #: Open event-channel subscriptions (T-EV): ``sub_id -> emit`` callback.
        self._subscriptions: dict[str, Callable[[dict[str, Any]], None]] = {}
        self._closed: bool = False

    async def call(self, envelope: dict[str, Any]) -> dict[str, Any]:
        """Ship a ``native_call`` frame and await the client's ``native_result``.

        Args:
            envelope: A ``native_call`` envelope carrying a ``call_id``.

        Returns:
            The ``native_result`` envelope posted back by the client.

        Raises:
            BrowserUnavailableError: If the bridge has been closed.
            NativeError: With code ``timeout`` if no result arrives in time.
        """
        if self._closed:
            raise BrowserUnavailableError("proxy bridge is closed")
        call_id = str(envelope["call_id"])
        loop = asyncio.get_running_loop()
        future: asyncio.Future[dict[str, Any]] = loop.create_future()
        self._pending[call_id] = future
        try:
            self.send_frame(envelope)
            if self.timeout is None:
                return await future
            try:
                return await asyncio.wait_for(future, self.timeout)
            except TimeoutError as exc:
                raise NativeError(
                    "timeout",
                    f"{envelope.get('capability', 'native call')} did not answer "
                    f"within {self.timeout}s",
                ) from exc
        finally:
            self._pending.pop(call_id, None)

    def resolve(self, call_id: str, payload: dict[str, Any]) -> bool:
        """Resolve a pending call with a ``native_result`` frame from the client.

        The server session calls this when a ``native_result`` frame arrives back
        over the transport.

        Args:
            call_id: The correlation id from the ``native_result`` frame.
            payload: The result envelope ``{"ok": ..., "value"/"error": ...}``.

        Returns:
            ``True`` if a matching pending future was resolved, else ``False``.
        """
        return resolve_native_result(call_id, payload, self._pending)

    async def subscribe(
        self,
        capability: str,
        args: dict[str, Any],
        emit: Callable[[dict[str, Any]], None],
    ) -> str:
        """Open an event-channel subscription and ship a ``native_subscribe`` frame.

        Args:
            capability: The dotted streaming capability name.
            args: JSON-able subscription arguments.
            emit: Callback the session invokes (via :meth:`deliver_event`) for each
                inbound ``native_event`` frame tagged with the returned id.

        Returns:
            The subscription id.

        Raises:
            BrowserUnavailableError: If the bridge has been closed.
        """
        if self._closed:
            raise BrowserUnavailableError("proxy bridge is closed")
        sub_id = _next_sub_id()
        self._subscriptions[sub_id] = emit
        self.send_frame(native_subscribe(capability, args, sub_id))
        return sub_id

    async def unsubscribe(self, sub_id: str) -> None:
        """Close a subscription and ship a ``native_unsubscribe`` frame.

        Args:
            sub_id: The id returned by :meth:`subscribe`.
        """
        self._subscriptions.pop(sub_id, None)
        if not self._closed:
            self.send_frame(native_unsubscribe(sub_id))

    def deliver_event(self, sub_id: str, payload: dict[str, Any]) -> bool:
        """Deliver an inbound ``native_event`` frame to its subscription (Mode B).

        The server session calls this when a ``native_event`` frame arrives. A
        terminal event (``done`` or ``error``) also drops the subscription.

        Args:
            sub_id: The subscription id from the ``native_event`` frame.
            payload: The event payload (``{"event"|"error"|"done": ...}``).

        Returns:
            ``True`` if a matching subscription received it, else ``False``.
        """
        delivered = resolve_native_event(sub_id, payload, self._subscriptions)
        if payload.get("done", False) or "error" in payload:
            self._subscriptions.pop(sub_id, None)
        return delivered

    def fail_pending(self, exc: BaseException) -> None:
        """Settle every in-flight call with ``exc`` (without closing the bridge).

        Lets the owner (e.g. a Mode-B session at teardown) fail outstanding calls
        with a domain-specific error — such as a transport-closed error — instead
        of the plain :class:`asyncio.CancelledError` that :meth:`close` raises.

        Args:
            exc: The exception to set on each not-yet-settled pending future.
        """
        for future in self._pending.values():
            if not future.done():
                future.set_exception(exc)
        self._pending.clear()
        self._end_subscriptions(str(exc) or "transport_closed")

    def close(self) -> None:
        """Close the bridge, cancel in-flight calls, and end all subscriptions."""
        self._closed = True
        for future in self._pending.values():
            if not future.done():
                future.cancel()
        self._pending.clear()
        self._end_subscriptions("transport_closed")

    def _end_subscriptions(self, error: str) -> None:
        """Terminate every open subscription so its ``native_events`` loop ends.

        Args:
            error: The error code delivered to each subscription's ``emit`` so the
                consuming iterator raises (instead of hanging) at teardown.
        """
        for emit in list(self._subscriptions.values()):
            emit({"error": error})
        self._subscriptions.clear()

call async

call(envelope: dict[str, Any]) -> dict[str, Any]

Ship a native_call frame and await the client's native_result.

Parameters:

Name	Type	Description	Default
envelope	dict[str, Any]	A native_call envelope carrying a call_id.	required

Returns:

Type	Description
dict[str, Any]	The native_result envelope posted back by the client.

Raises:

Type	Description
BrowserUnavailableError	If the bridge has been closed.
NativeError	With code timeout if no result arrives in time.

Source code in tempestweb/native/bridges.py

async def call(self, envelope: dict[str, Any]) -> dict[str, Any]:
    """Ship a ``native_call`` frame and await the client's ``native_result``.

    Args:
        envelope: A ``native_call`` envelope carrying a ``call_id``.

    Returns:
        The ``native_result`` envelope posted back by the client.

    Raises:
        BrowserUnavailableError: If the bridge has been closed.
        NativeError: With code ``timeout`` if no result arrives in time.
    """
    if self._closed:
        raise BrowserUnavailableError("proxy bridge is closed")
    call_id = str(envelope["call_id"])
    loop = asyncio.get_running_loop()
    future: asyncio.Future[dict[str, Any]] = loop.create_future()
    self._pending[call_id] = future
    try:
        self.send_frame(envelope)
        if self.timeout is None:
            return await future
        try:
            return await asyncio.wait_for(future, self.timeout)
        except TimeoutError as exc:
            raise NativeError(
                "timeout",
                f"{envelope.get('capability', 'native call')} did not answer "
                f"within {self.timeout}s",
            ) from exc
    finally:
        self._pending.pop(call_id, None)

resolve

resolve(call_id: str, payload: dict[str, Any]) -> bool

Resolve a pending call with a native_result frame from the client.

The server session calls this when a native_result frame arrives back over the transport.

Parameters:

Name	Type	Description	Default
call_id	str	The correlation id from the native_result frame.	required
payload	dict[str, Any]	The result envelope {"ok": ..., "value"/"error": ...}.	required

Returns:

Type	Description
bool	True if a matching pending future was resolved, else False.

Source code in tempestweb/native/bridges.py

def resolve(self, call_id: str, payload: dict[str, Any]) -> bool:
    """Resolve a pending call with a ``native_result`` frame from the client.

    The server session calls this when a ``native_result`` frame arrives back
    over the transport.

    Args:
        call_id: The correlation id from the ``native_result`` frame.
        payload: The result envelope ``{"ok": ..., "value"/"error": ...}``.

    Returns:
        ``True`` if a matching pending future was resolved, else ``False``.
    """
    return resolve_native_result(call_id, payload, self._pending)

subscribe async

subscribe(capability: str, args: dict[str, Any], emit: Callable[[dict[str, Any]], None]) -> str

Open an event-channel subscription and ship a native_subscribe frame.

Parameters:

Name	Type	Description	Default
capability	str	The dotted streaming capability name.	required
args	dict[str, Any]	JSON-able subscription arguments.	required
emit	Callable[[dict[str, Any]], None]	Callback the session invokes (via :meth:deliver_event) for each inbound native_event frame tagged with the returned id.	required

Returns:

Type	Description
str	The subscription id.

Raises:

Type	Description
BrowserUnavailableError	If the bridge has been closed.

Source code in tempestweb/native/bridges.py

async def subscribe(
    self,
    capability: str,
    args: dict[str, Any],
    emit: Callable[[dict[str, Any]], None],
) -> str:
    """Open an event-channel subscription and ship a ``native_subscribe`` frame.

    Args:
        capability: The dotted streaming capability name.
        args: JSON-able subscription arguments.
        emit: Callback the session invokes (via :meth:`deliver_event`) for each
            inbound ``native_event`` frame tagged with the returned id.

    Returns:
        The subscription id.

    Raises:
        BrowserUnavailableError: If the bridge has been closed.
    """
    if self._closed:
        raise BrowserUnavailableError("proxy bridge is closed")
    sub_id = _next_sub_id()
    self._subscriptions[sub_id] = emit
    self.send_frame(native_subscribe(capability, args, sub_id))
    return sub_id

unsubscribe async

unsubscribe(sub_id: str) -> None

Close a subscription and ship a native_unsubscribe frame.

Parameters:

Name	Type	Description	Default
sub_id	str	The id returned by :meth:subscribe.	required

Source code in tempestweb/native/bridges.py

async def unsubscribe(self, sub_id: str) -> None:
    """Close a subscription and ship a ``native_unsubscribe`` frame.

    Args:
        sub_id: The id returned by :meth:`subscribe`.
    """
    self._subscriptions.pop(sub_id, None)
    if not self._closed:
        self.send_frame(native_unsubscribe(sub_id))

deliver_event

deliver_event(sub_id: str, payload: dict[str, Any]) -> bool

Deliver an inbound native_event frame to its subscription (Mode B).

The server session calls this when a native_event frame arrives. A terminal event (done or error) also drops the subscription.

Parameters:

Name	Type	Description	Default
sub_id	str	The subscription id from the native_event frame.	required
payload	dict[str, Any]	The event payload ({"event"|"error"|"done": ...}).	required

Returns:

Type	Description
bool	True if a matching subscription received it, else False.

Source code in tempestweb/native/bridges.py

def deliver_event(self, sub_id: str, payload: dict[str, Any]) -> bool:
    """Deliver an inbound ``native_event`` frame to its subscription (Mode B).

    The server session calls this when a ``native_event`` frame arrives. A
    terminal event (``done`` or ``error``) also drops the subscription.

    Args:
        sub_id: The subscription id from the ``native_event`` frame.
        payload: The event payload (``{"event"|"error"|"done": ...}``).

    Returns:
        ``True`` if a matching subscription received it, else ``False``.
    """
    delivered = resolve_native_event(sub_id, payload, self._subscriptions)
    if payload.get("done", False) or "error" in payload:
        self._subscriptions.pop(sub_id, None)
    return delivered

fail_pending

fail_pending(exc: BaseException) -> None

Settle every in-flight call with exc (without closing the bridge).

Lets the owner (e.g. a Mode-B session at teardown) fail outstanding calls with a domain-specific error — such as a transport-closed error — instead of the plain :class:asyncio.CancelledError that :meth:close raises.

Parameters:

Name	Type	Description	Default
exc	BaseException	The exception to set on each not-yet-settled pending future.	required

Source code in tempestweb/native/bridges.py

def fail_pending(self, exc: BaseException) -> None:
    """Settle every in-flight call with ``exc`` (without closing the bridge).

    Lets the owner (e.g. a Mode-B session at teardown) fail outstanding calls
    with a domain-specific error — such as a transport-closed error — instead
    of the plain :class:`asyncio.CancelledError` that :meth:`close` raises.

    Args:
        exc: The exception to set on each not-yet-settled pending future.
    """
    for future in self._pending.values():
        if not future.done():
            future.set_exception(exc)
    self._pending.clear()
    self._end_subscriptions(str(exc) or "transport_closed")

close

close() -> None

Close the bridge, cancel in-flight calls, and end all subscriptions.

Source code in tempestweb/native/bridges.py

def close(self) -> None:
    """Close the bridge, cancel in-flight calls, and end all subscriptions."""
    self._closed = True
    for future in self._pending.values():
        if not future.done():
            future.cancel()
    self._pending.clear()
    self._end_subscriptions("transport_closed")

Photo

Bases: BaseModel

A captured photo returned by the browser.

Attributes:

Name	Type	Description
mime_type	str	The image MIME type (e.g. "image/jpeg", "image/png").
width	int	Frame width in pixels.
height	int	Frame height in pixels.
data_base64	str	The image bytes, base64-encoded (JSON-safe over the wire). Empty when the capture asked not to carry them — see ref.
ref	str	An opaque handle to the same bytes, still held by the client. Hand it to :mod:tempestweb.native.imaging to compress or transform the photo without moving the pixels across the bridge, and to :func:tempestweb.native.http.upload to send them straight to the server.

Source code in tempestweb/native/camera.py

class Photo(BaseModel):
    """A captured photo returned by the browser.

    Attributes:
        mime_type: The image MIME type (e.g. ``"image/jpeg"``, ``"image/png"``).
        width: Frame width in pixels.
        height: Frame height in pixels.
        data_base64: The image bytes, base64-encoded (JSON-safe over the wire).
            Empty when the capture asked not to carry them — see ``ref``.
        ref: An opaque handle to the same bytes, still held by the client. Hand
            it to :mod:`tempestweb.native.imaging` to compress or transform the
            photo **without moving the pixels across the bridge**, and to
            :func:`tempestweb.native.http.upload` to send them straight to the
            server.
    """

    model_config = ConfigDict(frozen=True)

    mime_type: str = "image/jpeg"
    width: int = 0
    height: int = 0
    data_base64: str = Field(default="", repr=False)
    ref: str = ""

    def to_bytes(self) -> bytes:
        """Decode the photo to raw bytes.

        Returns:
            The decoded image bytes.
        """
        return base64.b64decode(self.data_base64)

to_bytes

to_bytes() -> bytes

Decode the photo to raw bytes.

Returns:

Type	Description
bytes	The decoded image bytes.

Source code in tempestweb/native/camera.py

def to_bytes(self) -> bytes:
    """Decode the photo to raw bytes.

    Returns:
        The decoded image bytes.
    """
    return base64.b64decode(self.data_base64)

ClipboardImage dataclass

An image read from the system clipboard.

Attributes:

Name	Type	Description
data_base64	str	The image bytes, base64-encoded (JSON-safe over the wire).
mime_type	str	The image MIME type (e.g. "image/png").

Source code in tempestweb/native/clipboard.py

@dataclass(frozen=True)
class ClipboardImage:
    """An image read from the system clipboard.

    Attributes:
        data_base64: The image bytes, base64-encoded (JSON-safe over the wire).
        mime_type: The image MIME type (e.g. ``"image/png"``).
    """

    data_base64: str
    mime_type: str

Capability dataclass

One native capability's contract entry.

Attributes:

Name	Type	Description
name	str	The dotted capability name ("group.verb"), the dispatch key.
group	str	The namespace it belongs to ("http", "storage", …).
mode_c	bool	Whether the Mode C facade (client/transpile/native.js) exposes it — i.e. a transpiled, Python-free app can call it in-process.
streaming	bool	Whether it is a streaming capability served over the native event channel (T-EV) — many events per subscription — rather than a single-shot request/response call. Streaming capabilities register in the client's EVENT_HANDLERS (not HANDLERS) and, in Python, are consumed via :func:~tempestweb.native.native_events / async for rather than await.

Source code in tempestweb/native/contract.py

@dataclass(frozen=True)
class Capability:
    """One native capability's contract entry.

    Attributes:
        name: The dotted capability name (``"group.verb"``), the dispatch key.
        group: The namespace it belongs to (``"http"``, ``"storage"``, …).
        mode_c: Whether the Mode C facade (``client/transpile/native.js``) exposes
            it — i.e. a transpiled, Python-free app can call it in-process.
        streaming: Whether it is a **streaming** capability served over the native
            event channel (T-EV) — many events per subscription — rather than a
            single-shot request/response call. Streaming capabilities register in
            the client's ``EVENT_HANDLERS`` (not ``HANDLERS``) and, in Python, are
            consumed via :func:`~tempestweb.native.native_events` /
            ``async for`` rather than ``await``.
    """

    name: str
    group: str
    mode_c: bool
    streaming: bool

BrowserUnavailableError

Bases: RuntimeError

Raised when a native call is made with no :class:NativeBridge installed.

The capability modules always reach the browser through an installed bridge. Off-platform (a plain Python process, a unit test that forgot to install a bridge), there is no browser to call, so dispatch fails fast with this error instead of silently no-op-ing.

Source code in tempestweb/native/dispatch.py

class BrowserUnavailableError(RuntimeError):
    """Raised when a native call is made with no :class:`NativeBridge` installed.

    The capability modules always reach the browser through an installed bridge.
    Off-platform (a plain Python process, a unit test that forgot to install a
    bridge), there is no browser to call, so dispatch fails fast with this error
    instead of silently no-op-ing.
    """

EventBridge

Bases: Protocol

A :class:NativeBridge that also serves the native event channel (T-EV).

The request/response :meth:NativeBridge.call seam is single-shot. Streaming capabilities (geolocation.watch, sensors, network/visibility/orientation change, media/idle, cross-tab broadcast receive, ...) need many events per subscription over time, so a streaming bridge additionally implements :meth:subscribe/:meth:unsubscribe.

A subscription delivers events through the injected emit callback. Each emitted payload is one of {"event": <value>} (a data event), {"error": <code>, "message": <detail>} (a terminal failure), or {"done": true} (the stream ended normally). emit may be called from a non-loop thread; implementations forward to the loop safely.

Source code in tempestweb/native/dispatch.py

@runtime_checkable
class EventBridge(Protocol):
    """A :class:`NativeBridge` that also serves the native **event channel** (T-EV).

    The request/response :meth:`NativeBridge.call` seam is single-shot. Streaming
    capabilities (``geolocation.watch``, sensors, network/visibility/orientation
    change, media/idle, cross-tab broadcast receive, ...) need many events per
    subscription over time, so a streaming bridge additionally implements
    :meth:`subscribe`/:meth:`unsubscribe`.

    A subscription delivers events through the injected ``emit`` callback. Each
    emitted payload is one of ``{"event": <value>}`` (a data event),
    ``{"error": <code>, "message": <detail>}`` (a terminal failure), or
    ``{"done": true}`` (the stream ended normally). ``emit`` may be called from a
    non-loop thread; implementations forward to the loop safely.
    """

    async def subscribe(
        self,
        capability: str,
        args: dict[str, Any],
        emit: Callable[[dict[str, Any]], None],
    ) -> str:
        """Open a subscription and stream its events through ``emit``.

        Args:
            capability: The dotted streaming capability name (``"geolocation.watch"``).
            args: JSON-able subscription arguments.
            emit: Callback invoked once per event with an ``{"event"|"error"|"done"}``
                payload. Safe to call from any thread.

        Returns:
            The subscription id used to later :meth:`unsubscribe`.
        """
        ...

    async def unsubscribe(self, sub_id: str) -> None:
        """Close a subscription so the browser stops delivering its events.

        Args:
            sub_id: The id returned by :meth:`subscribe`. Unknown ids are ignored.
        """
        ...

subscribe async

subscribe(capability: str, args: dict[str, Any], emit: Callable[[dict[str, Any]], None]) -> str

Open a subscription and stream its events through emit.

Parameters:

Name	Type	Description	Default
capability	str	The dotted streaming capability name ("geolocation.watch").	required
args	dict[str, Any]	JSON-able subscription arguments.	required
emit	Callable[[dict[str, Any]], None]	Callback invoked once per event with an {"event"|"error"|"done"} payload. Safe to call from any thread.	required

Returns:

Type	Description
str	The subscription id used to later :meth:unsubscribe.

Source code in tempestweb/native/dispatch.py

async def subscribe(
    self,
    capability: str,
    args: dict[str, Any],
    emit: Callable[[dict[str, Any]], None],
) -> str:
    """Open a subscription and stream its events through ``emit``.

    Args:
        capability: The dotted streaming capability name (``"geolocation.watch"``).
        args: JSON-able subscription arguments.
        emit: Callback invoked once per event with an ``{"event"|"error"|"done"}``
            payload. Safe to call from any thread.

    Returns:
        The subscription id used to later :meth:`unsubscribe`.
    """
    ...

unsubscribe async

unsubscribe(sub_id: str) -> None

Close a subscription so the browser stops delivering its events.

Parameters:

Name	Type	Description	Default
sub_id	str	The id returned by :meth:subscribe. Unknown ids are ignored.	required

Source code in tempestweb/native/dispatch.py

async def unsubscribe(self, sub_id: str) -> None:
    """Close a subscription so the browser stops delivering its events.

    Args:
        sub_id: The id returned by :meth:`subscribe`. Unknown ids are ignored.
    """
    ...

NativeBridge

Bases: Protocol

The seam between a native capability and the browser's Web API.

A bridge is installed once per running app (Mode A or Mode B) via :func:install_bridge. The capability modules call :meth:call without knowing which concrete bridge backs them.

Implementations must be safe to drive from an asyncio event loop.

Source code in tempestweb/native/dispatch.py

@runtime_checkable
class NativeBridge(Protocol):
    """The seam between a native capability and the browser's Web API.

    A bridge is installed once per running app (Mode A or Mode B) via
    :func:`install_bridge`. The capability modules call :meth:`call` without
    knowing which concrete bridge backs them.

    Implementations must be safe to drive from an asyncio event loop.
    """

    async def call(self, envelope: dict[str, Any]) -> dict[str, Any]:
        """Deliver a ``native_call`` envelope and await its ``native_result``.

        Args:
            envelope: A ``native_call`` envelope carrying a ``call_id``,
                ``capability`` and ``args``.

        Returns:
            The result envelope ``{"ok": bool, "value"/"error": ...}`` as produced
            by ``client/native/*.js``.

        Raises:
            BrowserUnavailableError: If the browser channel is gone.
        """
        ...

call async

call(envelope: dict[str, Any]) -> dict[str, Any]

Deliver a native_call envelope and await its native_result.

Parameters:

Name	Type	Description	Default
envelope	dict[str, Any]	A native_call envelope carrying a call_id, capability and args.	required

Returns:

Type	Description
dict[str, Any]	The result envelope {"ok": bool, "value"/"error": ...} as produced
dict[str, Any]	by client/native/*.js.

Raises:

Type	Description
BrowserUnavailableError	If the browser channel is gone.

Source code in tempestweb/native/dispatch.py

async def call(self, envelope: dict[str, Any]) -> dict[str, Any]:
    """Deliver a ``native_call`` envelope and await its ``native_result``.

    Args:
        envelope: A ``native_call`` envelope carrying a ``call_id``,
            ``capability`` and ``args``.

    Returns:
        The result envelope ``{"ok": bool, "value"/"error": ...}`` as produced
        by ``client/native/*.js``.

    Raises:
        BrowserUnavailableError: If the browser channel is gone.
    """
    ...

NativeError

Bases: RuntimeError

A native Web-capability call failed in the browser.

Attributes:

Name	Type	Description
code	str	A short machine-readable error code (e.g. "permission_denied", "unavailable", "not_found", "insecure_context", "http_error", "timeout").

Source code in tempestweb/native/dispatch.py

class NativeError(RuntimeError):
    """A native Web-capability call failed in the browser.

    Attributes:
        code: A short machine-readable error code (e.g. ``"permission_denied"``,
            ``"unavailable"``, ``"not_found"``, ``"insecure_context"``,
            ``"http_error"``, ``"timeout"``).
    """

    def __init__(self, code: str, message: str = "") -> None:
        """Initialize the error.

        Args:
            code: The machine-readable error code.
            message: A human-readable detail (optional).
        """
        self.code: str = code
        super().__init__(f"{code}: {message}" if message else code)

PickedFile

Bases: BaseModel

A file chosen by the user via the native file picker.

Attributes:

Name	Type	Description
data_base64	str	The file bytes, base64-encoded (no data-URI prefix).
mime	str	The file's MIME type as reported by the browser.
name	str	The original file name.

Source code in tempestweb/native/file.py

class PickedFile(BaseModel):
    """A file chosen by the user via the native file picker.

    Attributes:
        data_base64: The file bytes, base64-encoded (no data-URI prefix).
        mime: The file's MIME type as reported by the browser.
        name: The original file name.
    """

    model_config = ConfigDict(frozen=True)

    data_base64: str = Field(default="", repr=False)
    mime: str = "application/octet-stream"
    name: str = ""

    def to_bytes(self) -> bytes:
        """Decode the picked file to raw bytes.

        Returns:
            The decoded file bytes.
        """
        return base64.b64decode(self.data_base64)

to_bytes

to_bytes() -> bytes

Decode the picked file to raw bytes.

Returns:

Type	Description
bytes	The decoded file bytes.

Source code in tempestweb/native/file.py

def to_bytes(self) -> bytes:
    """Decode the picked file to raw bytes.

    Returns:
        The decoded file bytes.
    """
    return base64.b64decode(self.data_base64)

SaveResult

Bases: BaseModel

The outcome of a :func:save call.

Attributes:

Name	Type	Description
method	str	How the file was delivered — "share" (Web Share API) or "download" (anchor download).
shared	bool	True when the file went through the Web Share API.

Source code in tempestweb/native/file.py

class SaveResult(BaseModel):
    """The outcome of a :func:`save` call.

    Attributes:
        method: How the file was delivered — ``"share"`` (Web Share API) or
            ``"download"`` (anchor download).
        shared: ``True`` when the file went through the Web Share API.
    """

    model_config = ConfigDict(frozen=True)

    method: str = "download"
    shared: bool = False

FileHandle dataclass

A handle to a file opened or created through the File System Access API.

Attributes:

Name	Type	Description
id	str	The opaque handle id; the client holds the live FileSystemFileHandle in a registry keyed by this id.
name	str	The file name (e.g. "report.pdf").
mime_type	str	The file MIME type, or "" when unknown.
data_base64	str	The file bytes, base64-encoded; "" for handles returned by :func:save_file (which creates an empty file).

Source code in tempestweb/native/filesystem.py

@dataclass(frozen=True)
class FileHandle:
    """A handle to a file opened or created through the File System Access API.

    Attributes:
        id: The opaque handle id; the client holds the live ``FileSystemFileHandle``
            in a registry keyed by this id.
        name: The file name (e.g. ``"report.pdf"``).
        mime_type: The file MIME type, or ``""`` when unknown.
        data_base64: The file bytes, base64-encoded; ``""`` for handles returned by
            :func:`save_file` (which creates an empty file).
    """

    id: str
    name: str
    mime_type: str
    data_base64: str

Position

Bases: BaseModel

A geographic position fix returned by the browser.

Mirrors GeolocationCoordinates: accuracy is always present, while altitude is None when the device cannot report it.

Attributes:

Name	Type	Description
latitude	float	Latitude in decimal degrees.
longitude	float	Longitude in decimal degrees.
accuracy	float	Horizontal accuracy radius in meters (0.0 if unknown).
altitude	float | None	Altitude in meters above the WGS84 ellipsoid, or None.

Source code in tempestweb/native/geolocation.py

class Position(BaseModel):
    """A geographic position fix returned by the browser.

    Mirrors ``GeolocationCoordinates``: ``accuracy`` is always present, while
    ``altitude`` is ``None`` when the device cannot report it.

    Attributes:
        latitude: Latitude in decimal degrees.
        longitude: Longitude in decimal degrees.
        accuracy: Horizontal accuracy radius in meters (``0.0`` if unknown).
        altitude: Altitude in meters above the WGS84 ellipsoid, or ``None``.
    """

    model_config = ConfigDict(frozen=True)

    latitude: float
    longitude: float
    accuracy: float = 0.0
    altitude: float | None = None

HttpResponse

Bases: BaseModel

A typed HTTP response returned by the browser fetch call.

Attributes:

Name	Type	Description
status	int	The HTTP status code.
ok	bool	Whether status is in the 2xx range (mirrors Response.ok).
headers	dict[str, str]	Response headers, lower-cased keys.
text	str	The response body decoded as text (empty string when absent).
json_body	Any	The parsed JSON body when the response was JSON, else None. Carried on the wire under the key "json" (the field is named json_body to avoid shadowing :meth:pydantic.BaseModel.json).

Source code in tempestweb/native/http.py

class HttpResponse(BaseModel):
    """A typed HTTP response returned by the browser ``fetch`` call.

    Attributes:
        status: The HTTP status code.
        ok: Whether ``status`` is in the 2xx range (mirrors ``Response.ok``).
        headers: Response headers, lower-cased keys.
        text: The response body decoded as text (empty string when absent).
        json_body: The parsed JSON body when the response was JSON, else ``None``.
            Carried on the wire under the key ``"json"`` (the field is named
            ``json_body`` to avoid shadowing :meth:`pydantic.BaseModel.json`).
    """

    model_config = ConfigDict(frozen=True, populate_by_name=True)

    status: int
    ok: bool
    headers: dict[str, str] = Field(default_factory=dict)
    text: str = ""
    json_body: Any = Field(default=None, alias="json")

RetryOptions

Bases: BaseModel

Retry / exponential-backoff policy for :func:request.

An unknown keyword is an error, not a silent no-op: this is a policy the developer writes by hand, so RetryOptions(backoff=0.5) naming a field that does not exist means the request runs with the defaults while the code reads as if it were configured. Every widget in the tree already refuses a kwarg it does not declare — a policy object gets the same answer. Payload models parsed from the browser keep ignoring extras, because there a new client key must not break an older Python.

Attributes:

Name	Type	Description
attempts	int	Total attempts including the first try. 1 disables retry.
base_delay	float	Seconds to wait before the first retry.
factor	float	Multiplier applied to the delay after each failed attempt.
max_delay	float	Upper bound on any single backoff delay, in seconds.
retry_statuses	frozenset[int]	HTTP status codes that should trigger a retry.

Source code in tempestweb/native/http.py

class RetryOptions(BaseModel):
    """Retry / exponential-backoff policy for :func:`request`.

    An unknown keyword is an error, not a silent no-op: this is a policy the
    developer writes by hand, so ``RetryOptions(backoff=0.5)`` naming a field
    that does not exist means the request runs with the defaults while the code
    reads as if it were configured. Every widget in the tree already refuses a
    kwarg it does not declare — a policy object gets the same answer. Payload
    models parsed *from* the browser keep ignoring extras, because there a new
    client key must not break an older Python.

    Attributes:
        attempts: Total attempts including the first try. ``1`` disables retry.
        base_delay: Seconds to wait before the first retry.
        factor: Multiplier applied to the delay after each failed attempt.
        max_delay: Upper bound on any single backoff delay, in seconds.
        retry_statuses: HTTP status codes that should trigger a retry.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    attempts: int = Field(default=3, ge=1)
    base_delay: float = Field(default=0.2, ge=0.0)
    factor: float = Field(default=2.0, ge=1.0)
    max_delay: float = Field(default=10.0, ge=0.0)
    retry_statuses: frozenset[int] = Field(default=_DEFAULT_RETRY_STATUSES)

    def delay_for(self, attempt_index: int) -> float:
        """Compute the backoff delay before the retry numbered ``attempt_index``.

        Args:
            attempt_index: Zero-based index of the *upcoming* retry (``0`` is the
                wait before the first retry, i.e. after attempt 1 failed).

        Returns:
            The capped exponential delay in seconds.
        """
        delay = self.base_delay * (self.factor**attempt_index)
        return min(delay, self.max_delay)

delay_for

delay_for(attempt_index: int) -> float

Compute the backoff delay before the retry numbered attempt_index.

Parameters:

Name	Type	Description	Default
attempt_index	int	Zero-based index of the upcoming retry (0 is the wait before the first retry, i.e. after attempt 1 failed).	required

Returns:

Type	Description
float	The capped exponential delay in seconds.

Source code in tempestweb/native/http.py

def delay_for(self, attempt_index: int) -> float:
    """Compute the backoff delay before the retry numbered ``attempt_index``.

    Args:
        attempt_index: Zero-based index of the *upcoming* retry (``0`` is the
            wait before the first retry, i.e. after attempt 1 failed).

    Returns:
        The capped exponential delay in seconds.
    """
    delay = self.base_delay * (self.factor**attempt_index)
    return min(delay, self.max_delay)

IdleState

Bases: BaseModel

A snapshot of the user's idle state reported by the Idle Detection API.

Attributes:

Name	Type	Description
user	str	The user idle state, "active" or "idle".
screen	str	The screen state, "locked" or "unlocked".

Source code in tempestweb/native/idle.py

class IdleState(BaseModel):
    """A snapshot of the user's idle state reported by the Idle Detection API.

    Attributes:
        user: The user idle state, ``"active"`` or ``"idle"``.
        screen: The screen state, ``"locked"`` or ``"unlocked"``.
    """

    model_config = ConfigDict(frozen=True)

    user: str
    screen: str

InstallState

Bases: BaseModel

The current PWA install state.

Attributes:

Name	Type	Description
can_install	bool	A deferred prompt is available to fire.
installed	bool	The app reports as installed (standalone / appinstalled).
method	str	How the user can install here — "native" (a prompt is available), "ios" (manual Share → Add to Home) or "manual" (e.g. Firefox desktop). Lets the view show a native button vs. a platform tutorial.

Source code in tempestweb/native/install.py

class InstallState(BaseModel):
    """The current PWA install state.

    Attributes:
        can_install: A deferred prompt is available to fire.
        installed: The app reports as installed (standalone / appinstalled).
        method: How the user can install here — ``"native"`` (a prompt is
            available), ``"ios"`` (manual Share → Add to Home) or ``"manual"``
            (e.g. Firefox desktop). Lets the view show a native button vs. a
            platform tutorial.
    """

    model_config = ConfigDict(frozen=True)

    can_install: bool = False
    installed: bool = False
    method: str = "manual"

MidiMessage

Bases: BaseModel

A MIDI message received from an input port (event channel / T-EV).

Attributes:

Name	Type	Description
input_id	str	The id of the input port the message arrived on.
data	list[int]	The MIDI message bytes (0-255 each).
timestamp	float	The high-resolution timestamp (ms) the message was received.

Source code in tempestweb/native/midi.py

class MidiMessage(BaseModel):
    """A MIDI message received from an input port (event channel / T-EV).

    Attributes:
        input_id: The id of the input port the message arrived on.
        data: The MIDI message bytes (0-255 each).
        timestamp: The high-resolution timestamp (ms) the message was received.
    """

    model_config = ConfigDict(frozen=True)

    input_id: str
    data: list[int]
    timestamp: float

MidiPorts dataclass

The MIDI input and output ports exposed by the Web MIDI API.

Attributes:

Name	Type	Description
inputs	list[dict[str, Any]]	The available input ports as JSON-able dicts (id, name, …).
outputs	list[dict[str, Any]]	The available output ports as JSON-able dicts; the client holds the live MIDIOutput objects keyed by their id.

Source code in tempestweb/native/midi.py

@dataclass(frozen=True)
class MidiPorts:
    """The MIDI input and output ports exposed by the Web MIDI API.

    Attributes:
        inputs: The available input ports as JSON-able dicts (id, name, …).
        outputs: The available output ports as JSON-able dicts; the client holds
            the live ``MIDIOutput`` objects keyed by their id.
    """

    inputs: list[dict[str, Any]]
    outputs: list[dict[str, Any]]

NetworkState dataclass

A snapshot of the browser's network conditions.

Attributes:

Name	Type	Description
online	bool	Whether the browser reports itself as online.
effective_type	str	The effective connection type ("slow-2g", "2g", "3g", "4g"), or "" when unknown.
downlink	float	Estimated downlink bandwidth in megabits per second.
rtt	int	Estimated round-trip time in milliseconds.
save_data	bool	Whether the user has requested reduced data usage.

Source code in tempestweb/native/network.py

@dataclass(frozen=True)
class NetworkState:
    """A snapshot of the browser's network conditions.

    Attributes:
        online: Whether the browser reports itself as online.
        effective_type: The effective connection type (``"slow-2g"``, ``"2g"``,
            ``"3g"``, ``"4g"``), or ``""`` when unknown.
        downlink: Estimated downlink bandwidth in megabits per second.
        rtt: Estimated round-trip time in milliseconds.
        save_data: Whether the user has requested reduced data usage.
    """

    online: bool
    effective_type: str
    downlink: float
    rtt: int
    save_data: bool

NdefMessage

Bases: BaseModel

One NDEF message read from a nearby tag.

Attributes:

Name	Type	Description
serial_number	str	The tag's serial number, or "" when unavailable.
records	list[dict[str, Any]]	The decoded NDEF records (browser-defined shape).

Source code in tempestweb/native/nfc.py

class NdefMessage(BaseModel):
    """One NDEF message read from a nearby tag.

    Attributes:
        serial_number: The tag's serial number, or ``""`` when unavailable.
        records: The decoded NDEF records (browser-defined shape).
    """

    model_config = ConfigDict(frozen=True)

    serial_number: str = ""
    records: list[dict[str, Any]] = []

NotificationPermission

Bases: StrEnum

The browser's notification permission state.

Mirrors the Web NotificationPermission enum.

Attributes:

Name	Type	Description
DEFAULT		The user has not yet chosen (notifications are not allowed yet).
GRANTED		The user allowed notifications.
DENIED		The user blocked notifications.

Source code in tempestweb/native/notifications.py

class NotificationPermission(StrEnum):
    """The browser's notification permission state.

    Mirrors the Web ``NotificationPermission`` enum.

    Attributes:
        DEFAULT: The user has not yet chosen (notifications are not allowed yet).
        GRANTED: The user allowed notifications.
        DENIED: The user blocked notifications.
    """

    DEFAULT = "default"
    GRANTED = "granted"
    DENIED = "denied"

PushState

Bases: BaseModel

WebPush support and current permission, reported without prompting.

Attributes:

Name	Type	Description
supported	bool	Whether WebPush (service worker + PushManager + Notification) is available in this context.
permission	str	The current notification permission ("granted"/"denied"/"default"/"unsupported").

Source code in tempestweb/native/notifications.py

class PushState(BaseModel):
    """WebPush support and current permission, reported without prompting.

    Attributes:
        supported: Whether WebPush (service worker + PushManager + Notification)
            is available in this context.
        permission: The current notification permission
            (``"granted"``/``"denied"``/``"default"``/``"unsupported"``).
    """

    model_config = ConfigDict(frozen=True)

    supported: bool = False
    permission: str = "unsupported"

Mutation

Bases: BaseModel

A queued offline mutation.

Attributes:

Name	Type	Description
id	str	The queue row's primary key.
owner	str	The owner scope the mutation belongs to.
idempotency_key	str	The stable key the server dedups replays on.
method	str	The HTTP method ("POST"/"PUT"/"PATCH"/"DELETE").
url	str	The target URL.
attempts	int	How many replay attempts have been made.
status	str	The row status ("pending"/"done"/"failed"/ "conflict").

Source code in tempestweb/native/offline.py

class Mutation(BaseModel):
    """A queued offline mutation.

    Attributes:
        id: The queue row's primary key.
        owner: The owner scope the mutation belongs to.
        idempotency_key: The stable key the server dedups replays on.
        method: The HTTP method (``"POST"``/``"PUT"``/``"PATCH"``/``"DELETE"``).
        url: The target URL.
        attempts: How many replay attempts have been made.
        status: The row status (``"pending"``/``"done"``/``"failed"``/
            ``"conflict"``).
    """

    model_config = ConfigDict(frozen=True)

    id: str
    owner: str
    idempotency_key: str
    method: str
    url: str
    attempts: int = 0
    status: str = "pending"

ReplayResult

Bases: BaseModel

The outcome of a queue replay.

Attributes:

Name	Type	Description
sent	int	How many mutations were accepted and removed.
remaining	int	How many mutations are still pending.
failed	int	How many mutations were dead-lettered this run (permanent client error, or transient attempts exhausted).
conflicts	int	How many mutations were moved to the conflict lane this run (the server returned 409).

Source code in tempestweb/native/offline.py

class ReplayResult(BaseModel):
    """The outcome of a queue replay.

    Attributes:
        sent: How many mutations were accepted and removed.
        remaining: How many mutations are still pending.
        failed: How many mutations were dead-lettered this run (permanent client
            error, or transient attempts exhausted).
        conflicts: How many mutations were moved to the conflict lane this run
            (the server returned ``409``).
    """

    model_config = ConfigDict(frozen=True)

    sent: int = 0
    remaining: int = 0
    failed: int = 0
    conflicts: int = 0

OnnxModel

Bases: BaseModel

A loaded onnxruntime-web session living on the JS side.

Attributes:

Name	Type	Description
session_id	str	Opaque id used to address the cached session on onnx.run.
input_names	list[str]	The model's input names, in declaration order.
output_names	list[str]	The model's output names, in declaration order.

Source code in tempestweb/native/onnx.py

class OnnxModel(BaseModel):
    """A loaded onnxruntime-web session living on the JS side.

    Attributes:
        session_id: Opaque id used to address the cached session on ``onnx.run``.
        input_names: The model's input names, in declaration order.
        output_names: The model's output names, in declaration order.
    """

    model_config = ConfigDict(frozen=True)

    session_id: str
    input_names: list[str] = Field(default_factory=list)
    output_names: list[str] = Field(default_factory=list)

    @property
    def input_name(self) -> str:
        """Name of the first (and usually only) input.

        Returns:
            The first input name.

        Raises:
            IndexError: If the model declares no inputs.
        """
        return self.input_names[0]

input_name property

input_name: str

Name of the first (and usually only) input.

Returns:

Type	Description
str	The first input name.

Raises:

Type	Description
IndexError	If the model declares no inputs.

Tensor

Bases: BaseModel

A dense tensor crossing the bridge as base64-encoded raw bytes.

Attributes:

Name	Type	Description
data_base64	str	The raw little-endian tensor bytes, base64-encoded.
dims	list[int]	The tensor shape (e.g. [1, 3, 640, 640]).
dtype	str	The element type as an onnxruntime-web type string ("float32", "int64", "uint8", ...).

Source code in tempestweb/native/onnx.py

class Tensor(BaseModel):
    """A dense tensor crossing the bridge as base64-encoded raw bytes.

    Attributes:
        data_base64: The raw little-endian tensor bytes, base64-encoded.
        dims: The tensor shape (e.g. ``[1, 3, 640, 640]``).
        dtype: The element type as an onnxruntime-web type string
            (``"float32"``, ``"int64"``, ``"uint8"``, ...).
    """

    model_config = ConfigDict(frozen=True)

    data_base64: str = Field(default="", repr=False)
    dims: list[int] = Field(default_factory=list)
    dtype: str = "float32"

OrientationState dataclass

The current screen orientation.

Attributes:

Name	Type	Description
type	str	The orientation type (e.g. "portrait-primary", "landscape-primary").
angle	int	The orientation angle in degrees (0, 90, 180, 270).

Source code in tempestweb/native/orientation.py

@dataclass(frozen=True)
class OrientationState:
    """The current screen orientation.

    Attributes:
        type: The orientation type (e.g. ``"portrait-primary"``,
            ``"landscape-primary"``).
        angle: The orientation angle in degrees (``0``, ``90``, ``180``, ``270``).
    """

    type: str
    angle: int

StorageEstimate dataclass

An estimate of the origin's storage usage and quota.

Attributes:

Name	Type	Description
usage	int	Bytes currently used by the origin.
quota	int	Total bytes available to the origin.

Source code in tempestweb/native/quota.py

@dataclass(frozen=True)
class StorageEstimate:
    """An estimate of the origin's storage usage and quota.

    Attributes:
        usage: Bytes currently used by the origin.
        quota: Total bytes available to the origin.
    """

    usage: int
    quota: int

Recording dataclass

A finalized media recording.

Attributes:

Name	Type	Description
data_base64	str	The recorded bytes, base64-encoded (JSON-safe over the wire).
mime_type	str	The recording MIME type (e.g. "audio/webm").
size	int	The recorded byte length.

Source code in tempestweb/native/recorder.py

@dataclass(frozen=True)
class Recording:
    """A finalized media recording.

    Attributes:
        data_base64: The recorded bytes, base64-encoded (JSON-safe over the wire).
        mime_type: The recording MIME type (e.g. ``"audio/webm"``).
        size: The recorded byte length.
    """

    data_base64: str
    mime_type: str
    size: int

DeviceOrientation

Bases: BaseModel

A device-orientation reading from the Device Orientation API.

Attributes:

Name	Type	Description
alpha	float | None	Rotation around the z-axis in degrees (0-360), or None when unavailable.
beta	float | None	Front-to-back tilt in degrees (-180-180), or None.
gamma	float | None	Left-to-right tilt in degrees (-90-90), or None.
absolute	bool	Whether the reading is relative to Earth's coordinate frame.

Source code in tempestweb/native/sensors.py

class DeviceOrientation(BaseModel):
    """A device-orientation reading from the Device Orientation API.

    Attributes:
        alpha: Rotation around the z-axis in degrees (0-360), or ``None`` when
            unavailable.
        beta: Front-to-back tilt in degrees (-180-180), or ``None``.
        gamma: Left-to-right tilt in degrees (-90-90), or ``None``.
        absolute: Whether the reading is relative to Earth's coordinate frame.
    """

    model_config = ConfigDict(frozen=True)

    alpha: float | None
    beta: float | None
    gamma: float | None
    absolute: bool

Motion

Bases: BaseModel

A device-motion reading from the Device Motion API.

Attributes:

Name	Type	Description
acceleration	dict[str, float | None]	Acceleration on x/y/z axes in m/s^2; each value is None when the axis cannot be reported.
rotation_rate	dict[str, float | None]	Rotation rate around alpha/beta/gamma axes in degrees per second; each value is None when unavailable.
interval	float	The sampling interval in milliseconds between readings.

Source code in tempestweb/native/sensors.py

class Motion(BaseModel):
    """A device-motion reading from the Device Motion API.

    Attributes:
        acceleration: Acceleration on ``x``/``y``/``z`` axes in m/s^2; each value is
            ``None`` when the axis cannot be reported.
        rotation_rate: Rotation rate around ``alpha``/``beta``/``gamma`` axes in
            degrees per second; each value is ``None`` when unavailable.
        interval: The sampling interval in milliseconds between readings.
    """

    model_config = ConfigDict(frozen=True)

    acceleration: dict[str, float | None]
    rotation_rate: dict[str, float | None]
    interval: float

ShareOutcome

Bases: StrEnum

The outcome of a :func:share call.

Attributes:

Name	Type	Description
SHARED		The OS share sheet completed (content was shared).
CANCELLED		The user dismissed the share sheet.
UNSUPPORTED		The Web Share API is unavailable in this browser.

Source code in tempestweb/native/share.py

class ShareOutcome(StrEnum):
    """The outcome of a :func:`share` call.

    Attributes:
        SHARED: The OS share sheet completed (content was shared).
        CANCELLED: The user dismissed the share sheet.
        UNSUPPORTED: The Web Share API is unavailable in this browser.
    """

    SHARED = "shared"
    CANCELLED = "cancelled"
    UNSUPPORTED = "unsupported"

ShareResult

Bases: BaseModel

The typed result of a :func:share call.

Attributes:

Name	Type	Description
outcome	ShareOutcome	The :class:ShareOutcome.

Source code in tempestweb/native/share.py

class ShareResult(BaseModel):
    """The typed result of a :func:`share` call.

    Attributes:
        outcome: The :class:`ShareOutcome`.
    """

    model_config = ConfigDict(frozen=True)

    outcome: ShareOutcome

SpeechResult

Bases: BaseModel

A speech-recognition (STT) result from the Web Speech API.

Attributes:

Name	Type	Description
transcript	str	The recognized text for this result.
is_final	bool	Whether this is a finalized result (True) or an interim, still-changing hypothesis (False).
confidence	float	The recognizer's confidence in the transcript, 0.0-1.0.

Source code in tempestweb/native/speech.py

class SpeechResult(BaseModel):
    """A speech-recognition (STT) result from the Web Speech API.

    Attributes:
        transcript: The recognized text for this result.
        is_final: Whether this is a finalized result (``True``) or an interim,
            still-changing hypothesis (``False``).
        confidence: The recognizer's confidence in the transcript, ``0.0``-``1.0``.
    """

    model_config = ConfigDict(frozen=True)

    transcript: str
    is_final: bool
    confidence: float

Voice dataclass

A speech-synthesis voice available in the browser.

Attributes:

Name	Type	Description
name	str	The human-readable voice name (e.g. "Google US English").
lang	str	The BCP-47 language tag the voice speaks (e.g. "en-US").
default	bool	Whether this is the browser's default voice.

Source code in tempestweb/native/speech.py

@dataclass(frozen=True)
class Voice:
    """A speech-synthesis voice available in the browser.

    Attributes:
        name: The human-readable voice name (e.g. ``"Google US English"``).
        lang: The BCP-47 language tag the voice speaks (e.g. ``"en-US"``).
        default: Whether this is the browser's default voice.
    """

    name: str
    lang: str
    default: bool

SyncState

Bases: BaseModel

The observable state of a sync source.

Attributes:

Name	Type	Description
phase	str	"idle" | "syncing" | "error".
online	bool	Last known connectivity.
pending	int	Pending (unpushed) mutation count.
last_synced_at	int | None	Epoch ms of the last successful sync, or None.
last_summary	SyncSummary | None	The last run's :class:SyncSummary, or None.
error	str | None	The last error message, or None.

Source code in tempestweb/native/sync.py

class SyncState(BaseModel):
    """The observable state of a sync source.

    Attributes:
        phase: ``"idle"`` | ``"syncing"`` | ``"error"``.
        online: Last known connectivity.
        pending: Pending (unpushed) mutation count.
        last_synced_at: Epoch ms of the last successful sync, or None.
        last_summary: The last run's :class:`SyncSummary`, or None.
        error: The last error message, or None.
    """

    model_config = ConfigDict(frozen=True)

    phase: str = "idle"
    online: bool = True
    pending: int = 0
    last_synced_at: int | None = None
    last_summary: SyncSummary | None = None
    error: str | None = None

SyncSummary

Bases: BaseModel

The outcome of one sync run.

Attributes:

Name	Type	Description
sent	int	Mutations accepted and removed from the write queue.
remaining	int	Mutations still pending upload.
failed	int	Mutations dead-lettered this run.
conflicts	int	Mutations moved to the conflict lane this run.
applied	int	Remote rows applied by the pull.

Source code in tempestweb/native/sync.py

class SyncSummary(BaseModel):
    """The outcome of one sync run.

    Attributes:
        sent: Mutations accepted and removed from the write queue.
        remaining: Mutations still pending upload.
        failed: Mutations dead-lettered this run.
        conflicts: Mutations moved to the conflict lane this run.
        applied: Remote rows applied by the pull.
    """

    model_config = ConfigDict(frozen=True)

    sent: int = 0
    remaining: int = 0
    failed: int = 0
    conflicts: int = 0
    applied: int = 0

UsbDevice dataclass

A USB device granted through the WebUSB API.

Attributes:

Name	Type	Description
id	str	The opaque device id; the client holds the live USBDevice in a registry keyed by this id.
vendor_id	int	The USB vendor id.
product_id	int	The USB product id.
product_name	str	The device's product name, or "" when unknown.

Source code in tempestweb/native/usb.py

@dataclass(frozen=True)
class UsbDevice:
    """A USB device granted through the WebUSB API.

    Attributes:
        id: The opaque device id; the client holds the live ``USBDevice`` in a
            registry keyed by this id.
        vendor_id: The USB vendor id.
        product_id: The USB product id.
        product_name: The device's product name, or ``""`` when unknown.
    """

    id: str
    vendor_id: int
    product_id: int
    product_name: str

Level

Bases: BaseModel

One analysis frame.

Attributes:

Name	Type	Description
rms	float	Loudness over the frame's samples, 0.0–1.0.
peak	float	The loudest single sample in the frame, 0.0–1.0.
bands	list[float]	The frequency bins averaged into buckets, each 0.0–1.0, low frequencies first.

Source code in tempestweb/native/webaudio.py

class Level(BaseModel):
    """One analysis frame.

    Attributes:
        rms: Loudness over the frame's samples, ``0.0``–``1.0``.
        peak: The loudest single sample in the frame, ``0.0``–``1.0``.
        bands: The frequency bins averaged into buckets, each ``0.0``–``1.0``,
            low frequencies first.
    """

    model_config = ConfigDict(frozen=True)

    rms: float = 0.0
    peak: float = 0.0
    bands: list[float] = Field(default_factory=list)

SequenceResult

Bases: BaseModel

What the client scheduled.

Attributes:

Name	Type	Description
scheduled	int	How many steps were scheduled.
ends_in_ms	int	When the last step ends, measured from the call.
blocked	bool	Whether the audio context is still suspended — the browser blocks audio until the first user gesture, and the phrase stays scheduled rather than raising, mirroring audio.play.

Source code in tempestweb/native/webaudio.py

class SequenceResult(BaseModel):
    """What the client scheduled.

    Attributes:
        scheduled: How many steps were scheduled.
        ends_in_ms: When the last step ends, measured from the call.
        blocked: Whether the audio context is still suspended — the browser blocks
            audio until the first user gesture, and the phrase stays scheduled
            rather than raising, mirroring ``audio.play``.
    """

    model_config = ConfigDict(frozen=True)

    scheduled: int = 0
    ends_in_ms: int = 0
    blocked: bool = False

Step

Bases: BaseModel

One note in a phrase.

Attributes:

Name	Type	Description
frequency	float	The pitch in hertz.
duration_ms	int	How long the note sounds.
start_ms	int	How long after the call the note starts — overlap two steps by giving them the same start_ms, which is how a chord is written.
type	str	The oscillator waveform ("sine", "square", "sawtooth", "triangle").
gain	float	The note's peak gain, from 0.0 (silent) to 1.0 (full).
attack_ms	int	Ramp-up from silence to gain. A note that jumps straight to full amplitude clicks, because the waveform starts mid-cycle.
release_ms	int	Ramp-down back to silence, for the same reason.

Source code in tempestweb/native/webaudio.py

class Step(BaseModel):
    """One note in a phrase.

    Attributes:
        frequency: The pitch in hertz.
        duration_ms: How long the note sounds.
        start_ms: How long after the call the note starts — overlap two steps by
            giving them the same ``start_ms``, which is how a chord is written.
        type: The oscillator waveform (``"sine"``, ``"square"``, ``"sawtooth"``,
            ``"triangle"``).
        gain: The note's peak gain, from ``0.0`` (silent) to ``1.0`` (full).
        attack_ms: Ramp-up from silence to ``gain``. A note that jumps straight to
            full amplitude clicks, because the waveform starts mid-cycle.
        release_ms: Ramp-down back to silence, for the same reason.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    frequency: float = Field(default=440.0, gt=0.0)
    duration_ms: int = Field(default=200, ge=0)
    start_ms: int = Field(default=0, ge=0)
    type: str = "sine"
    gain: float = Field(default=0.5, ge=0.0, le=1.0)
    attack_ms: int = Field(default=5, ge=0)
    release_ms: int = Field(default=40, ge=0)

capture async

capture(*, facing: str = 'environment', quality: float = 0.85, mime_type: str = 'image/jpeg', include_bytes: bool = True) -> Photo

Capture a single photo from the device camera.

Parameters:

Name	Type	Description	Default
facing	str	Preferred camera ("environment" rear, "user" front).	'environment'
quality	float	Encoding quality in [0.0, 1.0] for lossy formats.	0.85
mime_type	str	The desired output image MIME type.	'image/jpeg'
include_bytes	bool	Whether to carry the image bytes back to Python. Pass False when the photo is only going to be compressed and uploaded: the bytes then never cross the bridge at all, and :attr:Photo.ref addresses them where they already are. A 4 MB photo crosses as ~5.3 MB of base64, which in Mode B is a network trip.	True

Returns:

Type	Description
Photo	The captured :class:Photo.

Raises:

Type	Description
NativeError	If the user denies camera permission (permission_denied), no camera is present (unavailable), or the page is not a secure context (insecure_context).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/camera.py

async def capture(
    *,
    facing: str = "environment",
    quality: float = 0.85,
    mime_type: str = "image/jpeg",
    include_bytes: bool = True,
) -> Photo:
    """Capture a single photo from the device camera.

    Args:
        facing: Preferred camera (``"environment"`` rear, ``"user"`` front).
        quality: Encoding quality in ``[0.0, 1.0]`` for lossy formats.
        mime_type: The desired output image MIME type.
        include_bytes: Whether to carry the image bytes back to Python. Pass
            ``False`` when the photo is only going to be compressed and uploaded:
            the bytes then never cross the bridge at all, and
            :attr:`Photo.ref` addresses them where they already are. A 4 MB photo
            crosses as ~5.3 MB of base64, which in Mode B is a network trip.

    Returns:
        The captured :class:`Photo`.

    Raises:
        NativeError: If the user denies camera permission (``permission_denied``),
            no camera is present (``unavailable``), or the page is not a secure
            context (``insecure_context``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    clamped = max(0.0, min(1.0, quality))
    value = await send_native_call(
        "camera.capture",
        {
            "facing": facing,
            "quality": clamped,
            "mime_type": mime_type,
            "include_bytes": include_bytes,
        },
    )
    return Photo.model_validate(value)

read async

read() -> str

Read the current text from the system clipboard.

Returns:

Type	Description
str	The clipboard text, or "" if the clipboard is empty or non-text.

Raises:

Type	Description
NativeError	If the read is blocked (permission_denied) or the page is not a secure context (insecure_context).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/clipboard.py

async def read() -> str:
    """Read the current text from the system clipboard.

    Returns:
        The clipboard text, or ``""`` if the clipboard is empty or non-text.

    Raises:
        NativeError: If the read is blocked (``permission_denied``) or the page is
            not a secure context (``insecure_context``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("clipboard.read", {})
    return str(value.get("text", ""))

write async

write(text: str) -> None

Write text to the system clipboard.

Parameters:

Name	Type	Description	Default
text	str	The text to place on the clipboard.	required

Raises:

Type	Description
NativeError	If the write is blocked (permission_denied) or the page is not a secure context (insecure_context).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/clipboard.py

async def write(text: str) -> None:
    """Write text to the system clipboard.

    Args:
        text: The text to place on the clipboard.

    Raises:
        NativeError: If the write is blocked (``permission_denied``) or the page is
            not a secure context (``insecure_context``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    await send_native_call("clipboard.write", {"text": text})

capability_names

capability_names() -> frozenset[str]

Return the full set of dotted capability names.

Returns:

Type	Description
frozenset[str]	Every capability's name, as a set for order-agnostic comparison.

Source code in tempestweb/native/contract.py

def capability_names() -> frozenset[str]:
    """Return the full set of dotted capability names.

    Returns:
        Every capability's ``name``, as a set for order-agnostic comparison.
    """
    return frozenset(cap.name for cap in CAPABILITIES)

mode_c_capability_names

mode_c_capability_names() -> frozenset[str]

Return the set of dotted names the Mode C facade exposes.

Returns:

Type	Description
frozenset[str]	The mode_c capabilities' names, as a set.

Source code in tempestweb/native/contract.py

def mode_c_capability_names() -> frozenset[str]:
    """Return the set of dotted names the Mode C facade exposes.

    Returns:
        The ``mode_c`` capabilities' names, as a set.
    """
    return frozenset(cap.name for cap in MODE_C_CAPABILITIES)

streaming_capability_names

streaming_capability_names() -> frozenset[str]

Return the set of streaming (event-channel) capability names.

Returns:

Type	Description
frozenset[str]	The streaming capabilities' names, as a set. These register in the
frozenset[str]	client's EVENT_HANDLERS rather than HANDLERS.

Source code in tempestweb/native/contract.py

def streaming_capability_names() -> frozenset[str]:
    """Return the set of streaming (event-channel) capability names.

    Returns:
        The ``streaming`` capabilities' names, as a set. These register in the
        client's ``EVENT_HANDLERS`` rather than ``HANDLERS``.
    """
    return frozenset(cap.name for cap in STREAMING_CAPABILITIES)

current_bridge

current_bridge() -> NativeBridge

Return the bridge installed in the current context, raising if none.

Returns:

Type	Description
NativeBridge	The context-local :class:NativeBridge.

Raises:

Type	Description
BrowserUnavailableError	If no bridge has been installed in this context.

Source code in tempestweb/native/dispatch.py

def current_bridge() -> NativeBridge:
    """Return the bridge installed in the current context, raising if none.

    Returns:
        The context-local :class:`NativeBridge`.

    Raises:
        BrowserUnavailableError: If no bridge has been installed in this context.
    """
    bridge = _bridge.get()
    if bridge is None:
        raise BrowserUnavailableError(
            "no native bridge installed (off-platform, or bootstrap incomplete)"
        )
    return bridge

install_bridge

install_bridge(bridge: NativeBridge) -> None

Install the native bridge for the current execution mode and context.

Called once during app bootstrap — by the WASM runtime (Mode A) with an in-process FFI bridge, or by each server session (Mode B) with its own transport bridge. The bridge is stored in a context-local variable, so a Mode-B server serving many connections keeps each session's bridge isolated (the call must run in that connection's task — which it does, since the session's :meth:~tempestweb.runtime.session.AppSession.start is awaited from its own run task).

Parameters:

Name	Type	Description	Default
bridge	NativeBridge	The :class:NativeBridge implementation to route native calls through.	required

Source code in tempestweb/native/dispatch.py

def install_bridge(bridge: NativeBridge) -> None:
    """Install the native bridge for the current execution mode and context.

    Called once during app bootstrap — by the WASM runtime (Mode A) with an
    in-process FFI bridge, or by each server session (Mode B) with its own
    transport bridge. The bridge is stored in a context-local variable, so a
    Mode-B server serving many connections keeps each session's bridge isolated
    (the call must run in that connection's task — which it does, since the
    session's :meth:`~tempestweb.runtime.session.AppSession.start` is awaited
    from its own ``run`` task).

    Args:
        bridge: The :class:`NativeBridge` implementation to route native calls
            through.
    """
    _bridge.set(bridge)

native_call

native_call(capability: str, args: dict[str, Any], call_id: str) -> dict[str, Any]

Build a native_call envelope matching docs/contract.md.

Parameters:

Name	Type	Description	Default
capability	str	The stable dotted capability name (e.g. "geolocation.get", "http.request", "clipboard.read").	required
args	dict[str, Any]	JSON-able arguments for the capability.	required
call_id	str	The correlation id the client echoes back with the result.	required

Returns:

Type	Description
dict[str, Any]	The serializable native_call envelope.

Source code in tempestweb/native/dispatch.py

def native_call(capability: str, args: dict[str, Any], call_id: str) -> dict[str, Any]:
    """Build a ``native_call`` envelope matching ``docs/contract.md``.

    Args:
        capability: The stable dotted capability name (e.g. ``"geolocation.get"``,
            ``"http.request"``, ``"clipboard.read"``).
        args: JSON-able arguments for the capability.
        call_id: The correlation id the client echoes back with the result.

    Returns:
        The serializable ``native_call`` envelope.
    """
    return {
        "kind": "native_call",
        "call_id": call_id,
        "capability": capability,
        "args": args,
    }

native_events async

native_events(capability: str, args: dict[str, Any]) -> AsyncIterator[dict[str, Any]]

Subscribe to a streaming capability and yield its events (T-EV).

The plan-facing API for every streaming capability: it opens a subscription on the installed bridge, yields one dict per browser event, and guarantees the subscription is closed when the iterator is exhausted, broken out of, or its consumer is cancelled. Backpressure-free: events are buffered in an unbounded queue as the browser produces them.

Example::

async for pos in native_events("geolocation.watch", {"high_accuracy": True}):
    app.set_state(lambda s: setattr(s, "here", pos))

Parameters:

Name	Type	Description	Default
capability	str	The dotted streaming capability name.	required
args	dict[str, Any]	JSON-able subscription arguments.	required

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	Each event's value payload, as a dict.

Raises:

Type	Description
BrowserUnavailableError	If no bridge is installed, or the installed bridge does not support streaming.
NativeError	If the browser reports the subscription failed.

Source code in tempestweb/native/dispatch.py

async def native_events(
    capability: str, args: dict[str, Any]
) -> AsyncIterator[dict[str, Any]]:
    """Subscribe to a streaming capability and yield its events (T-EV).

    The plan-facing API for every streaming capability: it opens a subscription on
    the installed bridge, yields one ``dict`` per browser event, and guarantees the
    subscription is closed when the iterator is exhausted, broken out of, or its
    consumer is cancelled. Backpressure-free: events are buffered in an unbounded
    queue as the browser produces them.

    Example::

        async for pos in native_events("geolocation.watch", {"high_accuracy": True}):
            app.set_state(lambda s: setattr(s, "here", pos))

    Args:
        capability: The dotted streaming capability name.
        args: JSON-able subscription arguments.

    Yields:
        Each event's ``value`` payload, as a ``dict``.

    Raises:
        BrowserUnavailableError: If no bridge is installed, or the installed bridge
            does not support streaming.
        NativeError: If the browser reports the subscription failed.
    """
    bridge = current_bridge()
    if not isinstance(bridge, EventBridge):
        raise BrowserUnavailableError(
            "the installed native bridge does not support the event channel"
        )
    loop = asyncio.get_running_loop()
    queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()

    def emit(payload: dict[str, Any]) -> None:
        """Hand one streamed event to the async iterator draining the queue.

        The bridge may call this from outside the event loop's thread, so the
        enqueue is scheduled with ``call_soon_threadsafe`` rather than done
        directly — a plain ``put_nowait`` here would be a data race.

        Args:
            payload: The decoded ``native_event`` payload.
        """
        loop.call_soon_threadsafe(queue.put_nowait, payload)

    sub_id = await bridge.subscribe(capability, args, emit)
    try:
        while True:
            payload = await queue.get()
            if payload.get("done", False):
                return
            if "error" in payload:
                raise NativeError(
                    str(payload.get("error", "unknown")),
                    str(payload.get("message", "")),
                )
            event = payload.get("event", {})
            yield cast("dict[str, Any]", event) if isinstance(event, dict) else {}
    finally:
        await bridge.unsubscribe(sub_id)

native_subscribe

native_subscribe(capability: str, args: dict[str, Any], sub_id: str) -> dict[str, Any]

Build a native_subscribe envelope for the event channel (T-EV).

Parameters:

Name	Type	Description	Default
capability	str	The dotted streaming capability name ("geolocation.watch").	required
args	dict[str, Any]	JSON-able subscription arguments.	required
sub_id	str	The correlation id the client tags every event of this stream with.	required

Returns:

Type	Description
dict[str, Any]	The serializable native_subscribe envelope.

Source code in tempestweb/native/dispatch.py

def native_subscribe(
    capability: str, args: dict[str, Any], sub_id: str
) -> dict[str, Any]:
    """Build a ``native_subscribe`` envelope for the event channel (T-EV).

    Args:
        capability: The dotted streaming capability name (``"geolocation.watch"``).
        args: JSON-able subscription arguments.
        sub_id: The correlation id the client tags every event of this stream with.

    Returns:
        The serializable ``native_subscribe`` envelope.
    """
    return {
        "kind": "native_subscribe",
        "sub_id": sub_id,
        "capability": capability,
        "args": args,
    }

native_unsubscribe

native_unsubscribe(sub_id: str) -> dict[str, Any]

Build a native_unsubscribe envelope for the event channel (T-EV).

Parameters:

Name	Type	Description	Default
sub_id	str	The id of the subscription to close.	required

Returns:

Type	Description
dict[str, Any]	The serializable native_unsubscribe envelope.

Source code in tempestweb/native/dispatch.py

def native_unsubscribe(sub_id: str) -> dict[str, Any]:
    """Build a ``native_unsubscribe`` envelope for the event channel (T-EV).

    Args:
        sub_id: The id of the subscription to close.

    Returns:
        The serializable ``native_unsubscribe`` envelope.
    """
    return {"kind": "native_unsubscribe", "sub_id": sub_id}

resolve_native_event

resolve_native_event(sub_id: str, payload: dict[str, Any], subscriptions: dict[str, Callable[[dict[str, Any]], None]]) -> bool

Deliver an inbound native_event to its subscription's emit (Mode B).

Called (on the loop thread) by a Mode-B transport bridge when a native_event frame tagged with sub_id arrives. Mode A has no use for this — its FFI bridge invokes emit inline from the JS callback.

Parameters:

Name	Type	Description	Default
sub_id	str	The subscription id parsed from the native_event frame.	required
payload	dict[str, Any]	The event payload ({"event"|"error"|"done": ...}).	required
subscriptions	dict[str, Callable[[dict[str, Any]], None]]	The bridge's sub_id -> emit registry.	required

Returns:

Type	Description
bool	True if a matching subscription received the event, else False.

Source code in tempestweb/native/dispatch.py

def resolve_native_event(
    sub_id: str,
    payload: dict[str, Any],
    subscriptions: dict[str, Callable[[dict[str, Any]], None]],
) -> bool:
    """Deliver an inbound ``native_event`` to its subscription's ``emit`` (Mode B).

    Called (on the loop thread) by a Mode-B transport bridge when a
    ``native_event`` frame tagged with ``sub_id`` arrives. Mode A has no use for
    this — its FFI bridge invokes ``emit`` inline from the JS callback.

    Args:
        sub_id: The subscription id parsed from the ``native_event`` frame.
        payload: The event payload (``{"event"|"error"|"done": ...}``).
        subscriptions: The bridge's ``sub_id -> emit`` registry.

    Returns:
        ``True`` if a matching subscription received the event, else ``False``.
    """
    emit = subscriptions.get(sub_id)
    if emit is None:
        return False
    emit(payload)
    return True

resolve_native_result

resolve_native_result(call_id: str, payload: dict[str, Any], pending: dict[str, Future[dict[str, Any]]]) -> bool

Resolve a pending native call with the client's native_result (Mode B).

Called (on the loop thread) by a Mode-B transport bridge when a native_result envelope tagged with call_id arrives back over the channel. Mode A has no use for this — its FFI bridge resolves its own promise inline.

Parameters:

Name	Type	Description	Default
call_id	str	The correlation id parsed from the native_result envelope.	required
payload	dict[str, Any]	The result envelope ({"ok": ..., "value"/"error": ...}).	required
pending	dict[str, Future[dict[str, Any]]]	The bridge's call_id -> Future registry.	required

Returns:

Type	Description
bool	True if a matching pending future was resolved, False otherwise
bool	(unknown or already-settled id).

Source code in tempestweb/native/dispatch.py

def resolve_native_result(
    call_id: str,
    payload: dict[str, Any],
    pending: dict[str, asyncio.Future[dict[str, Any]]],
) -> bool:
    """Resolve a pending native call with the client's ``native_result`` (Mode B).

    Called (on the loop thread) by a Mode-B transport bridge when a
    ``native_result`` envelope tagged with ``call_id`` arrives back over the
    channel. Mode A has no use for this — its FFI bridge resolves its own promise
    inline.

    Args:
        call_id: The correlation id parsed from the ``native_result`` envelope.
        payload: The result envelope (``{"ok": ..., "value"/"error": ...}``).
        pending: The bridge's ``call_id -> Future`` registry.

    Returns:
        ``True`` if a matching pending future was resolved, ``False`` otherwise
        (unknown or already-settled id).
    """
    future = pending.get(call_id)
    if future is None or future.done():
        return False
    future.set_result(payload)
    return True

send_native_call async

send_native_call(capability: str, args: dict[str, Any]) -> dict[str, Any]

Send a native_call and await the browser's typed result.

Builds an envelope with a fresh call_id, hands it to the installed bridge, and unwraps the result: a successful value payload is returned; a failure (ok is false) is raised as :class:NativeError. Must be called from the asyncio loop the app runs on (i.e. inside a widget handler).

Parameters:

Name	Type	Description	Default
capability	str	The stable dotted capability name.	required
args	dict[str, Any]	JSON-able arguments for the capability.	required

Returns:

Type	Description
dict[str, Any]	The value payload of a successful result, as a dict.

Raises:

Type	Description
BrowserUnavailableError	If no bridge is installed (off-platform).
NativeError	If the browser reports the call failed (ok is false).

Source code in tempestweb/native/dispatch.py

async def send_native_call(capability: str, args: dict[str, Any]) -> dict[str, Any]:
    """Send a ``native_call`` and await the browser's typed result.

    Builds an envelope with a fresh ``call_id``, hands it to the installed bridge,
    and unwraps the result: a successful ``value`` payload is returned; a failure
    (``ok`` is false) is raised as :class:`NativeError`. Must be called from the
    asyncio loop the app runs on (i.e. inside a widget handler).

    Args:
        capability: The stable dotted capability name.
        args: JSON-able arguments for the capability.

    Returns:
        The ``value`` payload of a successful result, as a ``dict``.

    Raises:
        BrowserUnavailableError: If no bridge is installed (off-platform).
        NativeError: If the browser reports the call failed (``ok`` is false).
    """
    call_id = _next_call_id()
    result = await current_bridge().call(native_call(capability, args, call_id))
    if not result.get("ok", False):
        raise NativeError(
            str(result.get("error", "unknown")),
            str(result.get("message", "")),
        )
    value = result.get("value", {})
    return cast("dict[str, Any]", value) if isinstance(value, dict) else {}

uninstall_bridge

uninstall_bridge() -> None

Remove the installed bridge for the current context (off-platform state).

Used by tests and by session teardown so a stale bridge never leaks across apps. Resets the context-local bridge to None in the calling context.

Source code in tempestweb/native/dispatch.py

def uninstall_bridge() -> None:
    """Remove the installed bridge for the current context (off-platform state).

    Used by tests and by session teardown so a stale bridge never leaks across
    apps. Resets the context-local bridge to ``None`` in the calling context.
    """
    _bridge.set(None)

file_pick async

file_pick(*, accept: str = 'image/*', capture: str | None = None) -> PickedFile

Open a native file picker and return the chosen file's bytes.

The FilePicker widget's event carries only a uri/name, not bytes; this capability opens an <input type="file"> and reads the selection back as base64 — the gallery/upload path for an on-device pipeline.

Parameters:

Name	Type	Description	Default
accept	str	The accept filter (e.g. "image/*").	'image/*'
capture	str | None	Optional capture hint ("environment" / "user") to prefer the camera on mobile.	None

Returns:

Type	Description
PickedFile	The chosen :class:PickedFile.

Raises:

Type	Description
NativeError	If the user cancels (cancelled) or the read fails (read_failed).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/file.py

async def pick(*, accept: str = "image/*", capture: str | None = None) -> PickedFile:
    """Open a native file picker and return the chosen file's bytes.

    The FilePicker widget's event carries only a uri/name, not bytes; this
    capability opens an ``<input type="file">`` and reads the selection back as
    base64 — the gallery/upload path for an on-device pipeline.

    Args:
        accept: The accept filter (e.g. ``"image/*"``).
        capture: Optional capture hint (``"environment"`` / ``"user"``) to prefer
            the camera on mobile.

    Returns:
        The chosen :class:`PickedFile`.

    Raises:
        NativeError: If the user cancels (``cancelled``) or the read fails
            (``read_failed``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("file.pick", {"accept": accept, "capture": capture})
    return PickedFile.model_validate(value)

file_save async

file_save(filename: str, data: bytes, *, mime_type: str = 'application/octet-stream') -> SaveResult

Share or download a generated file in the browser.

Parameters:

Name	Type	Description	Default
filename	str	The suggested file name (e.g. "famacha-historico.zip").	required
data	bytes	The raw file bytes to deliver.	required
mime_type	str	The file's MIME type (e.g. "application/zip").	'application/octet-stream'

Returns:

Name	Type	Description
A	SaveResult	class:SaveResult describing how the file was delivered.

Raises:

Type	Description
NativeError	If the user cancels a share that cannot fall back (share_cancelled) or delivery otherwise fails.
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/file.py

async def save(
    filename: str,
    data: bytes,
    *,
    mime_type: str = "application/octet-stream",
) -> SaveResult:
    """Share or download a generated file in the browser.

    Args:
        filename: The suggested file name (e.g. ``"famacha-historico.zip"``).
        data: The raw file bytes to deliver.
        mime_type: The file's MIME type (e.g. ``"application/zip"``).

    Returns:
        A :class:`SaveResult` describing how the file was delivered.

    Raises:
        NativeError: If the user cancels a share that cannot fall back
            (``share_cancelled``) or delivery otherwise fails.
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call(
        "file.save",
        {
            "filename": filename,
            "data_base64": base64.b64encode(data).decode("ascii"),
            "mime": mime_type,
        },
    )
    return SaveResult.model_validate(value)

get_position async

get_position(high_accuracy: bool = True) -> Position

Request a single location fix from the browser.

Parameters:

Name	Type	Description	Default
high_accuracy	bool	Set enableHighAccuracy (prefer GPS) when True.	True

Returns:

Type	Description
Position	The current :class:Position.

Raises:

Type	Description
NativeError	If the user denies permission (permission_denied), the page is not a secure context (insecure_context), or no fix is available (unavailable).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/geolocation.py

async def get_position(high_accuracy: bool = True) -> Position:
    """Request a single location fix from the browser.

    Args:
        high_accuracy: Set ``enableHighAccuracy`` (prefer GPS) when ``True``.

    Returns:
        The current :class:`Position`.

    Raises:
        NativeError: If the user denies permission (``permission_denied``), the
            page is not a secure context (``insecure_context``), or no fix is
            available (``unavailable``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("geolocation.get", {"high_accuracy": high_accuracy})
    return Position.model_validate(value)

generate_idempotency_key

generate_idempotency_key() -> str

Generate a fresh, URL-safe idempotency key.

Mirrors the React SDK's generateIdempotencyKey. The key lets a retried (or offline-replayed) request be deduplicated server-side so its effect happens at most once.

Returns:

Type	Description
str	A random URL-safe token (32 hex-ish characters).

Source code in tempestweb/native/http.py

def generate_idempotency_key() -> str:
    """Generate a fresh, URL-safe idempotency key.

    Mirrors the React SDK's ``generateIdempotencyKey``. The key lets a retried (or
    offline-replayed) request be deduplicated server-side so its effect happens at
    most once.

    Returns:
        A random URL-safe token (32 hex-ish characters).
    """
    return secrets.token_urlsafe(24)

poll async

poll(url: str, *, until: Callable[[HttpResponse], bool], interval: float = 1.0, max_attempts: int = 30, headers: dict[str, str] | None = None, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep) -> HttpResponse

Poll a URL until a predicate is satisfied or attempts run out.

Parameters:

Name	Type	Description	Default
url	str	The URL to poll (always fetched with GET).	required
until	Callable[[HttpResponse], bool]	Predicate deciding when polling is done, given the latest response.	required
interval	float	Seconds to wait between polls.	1.0
max_attempts	int	Maximum number of polls before giving up.	30
headers	dict[str, str] | None	Extra request headers.	None
sleep	Callable[[float], Awaitable[None]]	Awaitable sleep, injected so tests can run without real delays.	sleep

Returns:

Name	Type	Description
The	HttpResponse	class:HttpResponse that first satisfied until.

Raises:

Type	Description
NativeError	If a poll fails at the network level, or the predicate is never satisfied within max_attempts (code="poll_exhausted").
BrowserUnavailableError	If no native bridge is installed.

Source code in tempestweb/native/http.py

async def poll(
    url: str,
    *,
    until: Callable[[HttpResponse], bool],
    interval: float = 1.0,
    max_attempts: int = 30,
    headers: dict[str, str] | None = None,
    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
) -> HttpResponse:
    """Poll a URL until a predicate is satisfied or attempts run out.

    Args:
        url: The URL to poll (always fetched with ``GET``).
        until: Predicate deciding when polling is done, given the latest response.
        interval: Seconds to wait between polls.
        max_attempts: Maximum number of polls before giving up.
        headers: Extra request headers.
        sleep: Awaitable sleep, injected so tests can run without real delays.

    Returns:
        The :class:`HttpResponse` that first satisfied ``until``.

    Raises:
        NativeError: If a poll fails at the network level, or the predicate is
            never satisfied within ``max_attempts`` (``code="poll_exhausted"``).
        BrowserUnavailableError: If no native bridge is installed.
    """
    response: HttpResponse | None = None
    for attempt in range(max_attempts):
        response = await request("GET", url, headers=headers)
        if until(response):
            return response
        if attempt < max_attempts - 1:
            await sleep(interval)
    raise NativeError(
        "poll_exhausted",
        f"predicate not satisfied after {max_attempts} attempts polling {url}",
    )

request async

request(method: str, url: str, *, json: Any = None, headers: dict[str, str] | None = None, retry: RetryOptions | None = None, idempotency_key: str | None = None, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep) -> HttpResponse

Perform an HTTP request with optional retry and idempotency.

The request is retried with exponential backoff when it fails transiently (a retryable status code, or a network-level :class:NativeError) and the request is safe to retry — an idempotent method, or any method carrying an idempotency_key. The key is sent as the Idempotency-Key header so the server deduplicates the effect across retries and offline replays.

Parameters:

Name	Type	Description	Default
method	str	The HTTP method (case-insensitive).	required
url	str	The request URL.	required
json	Any	A JSON-able request body, or None.	None
headers	dict[str, str] | None	Extra request headers.	None
retry	RetryOptions | None	The retry policy. None means no retry (a single attempt).	None
idempotency_key	str | None	An explicit idempotency key; also makes a non-idempotent method (e.g. POST) eligible for retry.	None
sleep	Callable[[float], Awaitable[None]]	Awaitable sleep, injected so tests can run without real delays.	sleep

Returns:

Type	Description
HttpResponse	The final :class:HttpResponse — the first success, or the last response
HttpResponse	after exhausting retries.

Raises:

Type	Description
NativeError	If every attempt fails at the network level.
BrowserUnavailableError	If no native bridge is installed.

Source code in tempestweb/native/http.py

async def request(
    method: str,
    url: str,
    *,
    json: Any = None,  # noqa: ANN401 — a JSON request body is any JSON-able value
    headers: dict[str, str] | None = None,
    retry: RetryOptions | None = None,
    idempotency_key: str | None = None,
    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
) -> HttpResponse:
    """Perform an HTTP request with optional retry and idempotency.

    The request is retried with exponential backoff when it fails transiently
    (a retryable status code, or a network-level :class:`NativeError`) **and** the
    request is safe to retry — an idempotent method, or any method carrying an
    ``idempotency_key``. The key is sent as the ``Idempotency-Key`` header so the
    server deduplicates the effect across retries and offline replays.

    Args:
        method: The HTTP method (case-insensitive).
        url: The request URL.
        json: A JSON-able request body, or ``None``.
        headers: Extra request headers.
        retry: The retry policy. ``None`` means no retry (a single attempt).
        idempotency_key: An explicit idempotency key; also makes a non-idempotent
            method (e.g. ``POST``) eligible for retry.
        sleep: Awaitable sleep, injected so tests can run without real delays.

    Returns:
        The final :class:`HttpResponse` — the first success, or the last response
        after exhausting retries.

    Raises:
        NativeError: If every attempt fails at the network level.
        BrowserUnavailableError: If no native bridge is installed.
    """
    method_upper = method.upper()
    merged_headers: dict[str, str] = dict(headers or {})
    if idempotency_key is not None:
        merged_headers.setdefault("Idempotency-Key", idempotency_key)

    policy = retry or RetryOptions(attempts=1)
    retryable = method_upper in IDEMPOTENT_METHODS or idempotency_key is not None

    last_error: NativeError | None = None
    last_response: HttpResponse | None = None
    for attempt in range(policy.attempts):
        try:
            response = await _dispatch_request(
                method_upper, url, json_body=json, headers=merged_headers
            )
        except NativeError as exc:
            last_error = exc
            last_response = None
            if not retryable or attempt == policy.attempts - 1:
                raise
        else:
            last_response = response
            last_error = None
            if response.ok or response.status not in policy.retry_statuses:
                return response
            if not retryable or attempt == policy.attempts - 1:
                return response
        await sleep(policy.delay_for(attempt))

    # Unreachable in practice (the loop always returns or raises on the last
    # attempt); kept for type-completeness.
    if last_error is not None:
        raise last_error
    assert last_response is not None
    return last_response

upload async

upload(url: str, file: dict[str, Any], *, headers: dict[str, str] | None = None, on_progress: Callable[[float], None] | None = None) -> HttpResponse

Upload a file, reporting progress via a callback.

The browser performs a streaming upload (XMLHttpRequest/fetch with an upload progress listener); each progress tick is proxied back and forwarded to on_progress as a fraction in [0.0, 1.0].

Parameters:

Name	Type	Description	Default
url	str	The upload endpoint.	required
file	dict[str, Any]	A JSON-able descriptor of the file to upload, e.g. {"name": "a.png", "type": "image/png", "data": "<base64>"} or a client-side blob reference {"name": ..., "blob_id": ...}.	required
headers	dict[str, str] | None	Extra request headers.	None
on_progress	Callable[[float], None] | None	Optional callback receiving the upload fraction. The final tick is always 1.0 on success.	None

Returns:

Name	Type	Description
The	HttpResponse	class:HttpResponse for the completed upload.

Raises:

Type	Description
NativeError	If the upload fails.
BrowserUnavailableError	If no native bridge is installed.

Source code in tempestweb/native/http.py

async def upload(
    url: str,
    file: dict[str, Any],
    *,
    headers: dict[str, str] | None = None,
    on_progress: Callable[[float], None] | None = None,
) -> HttpResponse:
    """Upload a file, reporting progress via a callback.

    The browser performs a streaming upload (``XMLHttpRequest``/``fetch`` with an
    upload progress listener); each progress tick is proxied back and forwarded to
    ``on_progress`` as a fraction in ``[0.0, 1.0]``.

    Args:
        url: The upload endpoint.
        file: A JSON-able descriptor of the file to upload, e.g.
            ``{"name": "a.png", "type": "image/png", "data": "<base64>"}`` or a
            client-side blob reference ``{"name": ..., "blob_id": ...}``.
        headers: Extra request headers.
        on_progress: Optional callback receiving the upload fraction. The final
            tick is always ``1.0`` on success.

    Returns:
        The :class:`HttpResponse` for the completed upload.

    Raises:
        NativeError: If the upload fails.
        BrowserUnavailableError: If no native bridge is installed.
    """
    value = await send_native_call(
        "http.upload",
        {"url": url, "file": file, "headers": dict(headers or {})},
    )
    if on_progress is not None:
        ticks = value.get("progress", [])
        if isinstance(ticks, list):
            for tick in ticks:
                on_progress(float(tick))
        on_progress(1.0)
    return HttpResponse.model_validate(value.get("response", value))

install_prompt async

install_prompt() -> str

Fire the stashed native install prompt after a user gesture.

Returns:

Type	Description
str	The outcome: "accepted", "dismissed", or "unavailable" (no
str	prompt was captured, or it was already used).

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/install.py

async def prompt() -> str:
    """Fire the stashed native install prompt after a user gesture.

    Returns:
        The outcome: ``"accepted"``, ``"dismissed"``, or ``"unavailable"`` (no
        prompt was captured, or it was already used).

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("install.prompt", {})
    outcome = value.get("outcome", "unavailable")
    return str(outcome)

install_state async

install_state() -> InstallState

Report whether the app is installable and/or already installed.

Returns:

Type	Description
InstallState	The current :class:InstallState.

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/install.py

async def state() -> InstallState:
    """Report whether the app is installable and/or already installed.

    Returns:
        The current :class:`InstallState`.

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("install.state", {})
    return InstallState.model_validate(value)

notify async

notify(title: str, body: str = '') -> None

Post a local system notification.

The notification only appears if permission has been granted (see :func:request_permission); otherwise the browser silently drops it.

Parameters:

Name	Type	Description	Default
title	str	The notification title.	required
body	str	The notification body text.	''

Raises:

Type	Description
NativeError	If the Notifications API is unavailable (unavailable).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/notifications.py

async def notify(title: str, body: str = "") -> None:
    """Post a local system notification.

    The notification only appears if permission has been granted (see
    :func:`request_permission`); otherwise the browser silently drops it.

    Args:
        title: The notification title.
        body: The notification body text.

    Raises:
        NativeError: If the Notifications API is unavailable (``unavailable``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    await send_native_call("notifications.notify", {"title": title, "body": body})

push_state async

push_state() -> PushState

Report WebPush support and current permission WITHOUT prompting.

Use this to decide whether to show an "enable notifications" button before calling :func:subscribe (which must follow a user gesture).

Returns:

Name	Type	Description
The	PushState	class:PushState (support flag + current permission).

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/notifications.py

async def push_state() -> PushState:
    """Report WebPush support and current permission WITHOUT prompting.

    Use this to decide whether to show an "enable notifications" button before
    calling :func:`subscribe` (which must follow a user gesture).

    Returns:
        The :class:`PushState` (support flag + current permission).

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("notifications.push_state", {})
    return PushState.model_validate(value)

request_permission async

request_permission() -> NotificationPermission

Request permission to show notifications, awaiting the user's choice.

Returns:

Type	Description
NotificationPermission	The resulting :class:NotificationPermission after the prompt (or the
NotificationPermission	existing state if the user already chose).

Raises:

Type	Description
NativeError	If the Notifications API is unavailable (unavailable).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/notifications.py

async def request_permission() -> NotificationPermission:
    """Request permission to show notifications, awaiting the user's choice.

    Returns:
        The resulting :class:`NotificationPermission` after the prompt (or the
        existing state if the user already chose).

    Raises:
        NativeError: If the Notifications API is unavailable (``unavailable``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("notifications.request_permission", {})
    return NotificationPermission(str(value.get("permission", "default")))

subscribe async

subscribe(vapid_public_key: str) -> dict[str, Any]

Subscribe to WebPush, returning the raw browser subscription (P3).

Asks the client to run the browser-side push flow (ensure permission, create or reuse the pushManager subscription) with the given VAPID public key, and hands the subscription JSON back. Persist it server-side however your app likes (the framework does not own the endpoint schema).

Parameters:

Name	Type	Description	Default
vapid_public_key	str	The base64url-encoded VAPID application server key.	required

Returns:

Type	Description
dict[str, Any]	The push subscription as a JSON-able dict (endpoint, keys, ...).

Raises:

Type	Description
NativeError	If push is unsupported, permission is denied, or no service worker registration is available.
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/notifications.py

async def subscribe(vapid_public_key: str) -> dict[str, Any]:
    """Subscribe to WebPush, returning the raw browser subscription (P3).

    Asks the client to run the browser-side push flow (ensure permission, create
    or reuse the ``pushManager`` subscription) with the given VAPID public key, and
    hands the subscription JSON back. Persist it server-side however your app likes
    (the framework does not own the endpoint schema).

    Args:
        vapid_public_key: The base64url-encoded VAPID application server key.

    Returns:
        The push subscription as a JSON-able dict (``endpoint``, ``keys``, ...).

    Raises:
        NativeError: If push is unsupported, permission is denied, or no service
            worker registration is available.
        BrowserUnavailableError: If called with no native bridge installed.
    """
    return await send_native_call(
        "notifications.subscribe", {"vapid_public_key": vapid_public_key}
    )

unsubscribe async

unsubscribe() -> bool

Cancel the current WebPush subscription, if any (P3).

Asks the client to unsubscribe from pushManager. Returns whether a subscription was actually cancelled (False when none existed).

Returns:

Type	Description
bool	True if a subscription was cancelled, False otherwise.

Raises:

Type	Description
NativeError	If the unsubscribe call fails in the browser.
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/notifications.py

async def unsubscribe() -> bool:
    """Cancel the current WebPush subscription, if any (P3).

    Asks the client to unsubscribe from ``pushManager``. Returns whether a
    subscription was actually cancelled (``False`` when none existed).

    Returns:
        ``True`` if a subscription was cancelled, ``False`` otherwise.

    Raises:
        NativeError: If the unsubscribe call fails in the browser.
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("notifications.unsubscribe", {})
    return bool(value.get("unsubscribed", False))

offline_conflicts async

offline_conflicts(owner: str | None = None) -> list[Mutation]

List the mutations parked in the conflict lane for an owner.

A mutation lands here when the server rejects its replay with 409, signalling a write conflict that last-write-wins dedup cannot resolve. The row is kept (not dropped) so the app can reconcile it explicitly.

Parameters:

Name	Type	Description	Default
owner	str | None	The owner scope; defaults to the queue's default owner.	None

Returns:

Type	Description
list[Mutation]	The conflicting mutations (an empty list when none conflicted).

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/offline.py

async def conflicts(owner: str | None = None) -> list[Mutation]:
    """List the mutations parked in the conflict lane for an owner.

    A mutation lands here when the server rejects its replay with ``409``,
    signalling a write conflict that last-write-wins dedup cannot resolve. The
    row is kept (not dropped) so the app can reconcile it explicitly.

    Args:
        owner: The owner scope; defaults to the queue's default owner.

    Returns:
        The conflicting mutations (an empty list when none conflicted).

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("offline.conflicts", {"owner": owner})
    return [Mutation.model_validate(m) for m in value.get("mutations", [])]

offline_enqueue async

offline_enqueue(method: str, url: str, body: Any = None, *, idempotency_key: str | None = None, owner: str | None = None) -> Mutation

Enqueue a mutation for durable, replay-on-reconnect delivery.

Parameters:

Name	Type	Description	Default
method	str	The HTTP method.	required
url	str	The target URL.	required
body	Any	The JSON-able request body.	None
idempotency_key	str | None	An explicit key the server dedups on; generated when omitted.	None
owner	str | None	The owner scope; defaults to the queue's default owner.	None

Returns:

Type	Description
Mutation	The enqueued :class:Mutation.

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/offline.py

async def enqueue(
    method: str,
    url: str,
    body: Any = None,  # noqa: ANN401 - JSON-able request body of any shape
    *,
    idempotency_key: str | None = None,
    owner: str | None = None,
) -> Mutation:
    """Enqueue a mutation for durable, replay-on-reconnect delivery.

    Args:
        method: The HTTP method.
        url: The target URL.
        body: The JSON-able request body.
        idempotency_key: An explicit key the server dedups on; generated when
            omitted.
        owner: The owner scope; defaults to the queue's default owner.

    Returns:
        The enqueued :class:`Mutation`.

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call(
        "offline.enqueue",
        {
            "method": method,
            "url": url,
            "body": body,
            "idempotency_key": idempotency_key,
            "owner": owner,
        },
    )
    return Mutation.model_validate(value)

offline_failed async

offline_failed(owner: str | None = None) -> list[Mutation]

List the dead-lettered (permanently failed) mutations for an owner.

A mutation is dead-lettered when it hits a permanent client error (a non-retryable 4xx) or exhausts its retry attempts on a transient failure. These rows no longer block the queue and are surfaced here for inspection or a manual retry.

Parameters:

Name	Type	Description	Default
owner	str | None	The owner scope; defaults to the queue's default owner.	None

Returns:

Type	Description
list[Mutation]	The failed mutations (an empty list when none have failed).

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/offline.py

async def failed(owner: str | None = None) -> list[Mutation]:
    """List the dead-lettered (permanently failed) mutations for an owner.

    A mutation is dead-lettered when it hits a permanent client error (a
    non-retryable 4xx) or exhausts its retry attempts on a transient failure.
    These rows no longer block the queue and are surfaced here for inspection or
    a manual retry.

    Args:
        owner: The owner scope; defaults to the queue's default owner.

    Returns:
        The failed mutations (an empty list when none have failed).

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("offline.failed", {"owner": owner})
    return [Mutation.model_validate(m) for m in value.get("mutations", [])]

offline_pending async

offline_pending(owner: str | None = None) -> list[Mutation]

List the pending mutations for an owner, oldest first.

Parameters:

Name	Type	Description	Default
owner	str | None	The owner scope; defaults to the queue's default owner.	None

Returns:

Type	Description
list[Mutation]	The pending mutations (an empty list when none are queued).

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/offline.py

async def pending(owner: str | None = None) -> list[Mutation]:
    """List the pending mutations for an owner, oldest first.

    Args:
        owner: The owner scope; defaults to the queue's default owner.

    Returns:
        The pending mutations (an empty list when none are queued).

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("offline.pending", {"owner": owner})
    return [Mutation.model_validate(m) for m in value.get("mutations", [])]

offline_replay async

offline_replay(owner: str | None = None) -> ReplayResult

Replay the pending queue now.

Drains in FIFO order: accepted rows are removed; a transient failure stops replay to preserve order and is dead-lettered once attempts are exhausted; a permanent 4xx is dead-lettered immediately; a 409 moves the row to the conflict lane. Neither a dead-letter nor a conflict blocks the rest of the queue.

Parameters:

Name	Type	Description	Default
owner	str | None	The owner scope; defaults to the queue's default owner.	None

Returns:

Name	Type	Description
The	ReplayResult	class:ReplayResult (sent, remaining, failed and conflict counts).

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/offline.py

async def replay(owner: str | None = None) -> ReplayResult:
    """Replay the pending queue now.

    Drains in FIFO order: accepted rows are removed; a transient failure stops
    replay to preserve order and is dead-lettered once attempts are exhausted; a
    permanent 4xx is dead-lettered immediately; a ``409`` moves the row to the
    conflict lane. Neither a dead-letter nor a conflict blocks the rest of the
    queue.

    Args:
        owner: The owner scope; defaults to the queue's default owner.

    Returns:
        The :class:`ReplayResult` (sent, remaining, failed and conflict counts).

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("offline.replay", {"owner": owner})
    return ReplayResult.model_validate(value)

offline_size async

offline_size(owner: str | None = None) -> int

Count the pending mutations for an owner.

Parameters:

Name	Type	Description	Default
owner	str | None	The owner scope; defaults to the queue's default owner.	None

Returns:

Type	Description
int	The number of pending mutations.

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/offline.py

async def size(owner: str | None = None) -> int:
    """Count the pending mutations for an owner.

    Args:
        owner: The owner scope; defaults to the queue's default owner.

    Returns:
        The number of pending mutations.

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("offline.size", {"owner": owner})
    return int(value.get("size", 0))

is_share_supported async

is_share_supported() -> bool

Report whether the Web Share API is available in the current browser.

Returns:

Type	Description
bool	True if navigator.share exists (and the context permits sharing),
bool	False otherwise.

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/share.py

async def is_share_supported() -> bool:
    """Report whether the Web Share API is available in the current browser.

    Returns:
        ``True`` if ``navigator.share`` exists (and the context permits sharing),
        ``False`` otherwise.

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("share.is_supported", {})
    return bool(value.get("supported", False))

storage_get async

storage_get(name: str) -> str

Read the string value stored under a key.

Parameters:

Name	Type	Description	Default
name	str	The storage key (an IndexedDB key, scoped to the origin).	required

Returns:

Type	Description
str	The stored string value.

Raises:

Type	Description
NativeError	If the key does not exist (not_found).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/storage.py

async def get(name: str) -> str:
    """Read the string value stored under a key.

    Args:
        name: The storage key (an IndexedDB key, scoped to the origin).

    Returns:
        The stored string value.

    Raises:
        NativeError: If the key does not exist (``not_found``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("storage.get", {"name": name})
    return str(value.get("content", ""))

list_keys async

list_keys() -> list[str]

List the keys currently present in storage.

The keys are the origin's, not one owner's: on a device where two owners used the app, both sets come back.

Returns:

Type	Description
list[str]	The storage keys, or [] when storage is empty.

Raises:

Type	Description
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/storage.py

async def list_keys() -> list[str]:
    """List the keys currently present in storage.

    The keys are the origin's, not one owner's: on a device where two owners used
    the app, both sets come back.

    Returns:
        The storage keys, or ``[]`` when storage is empty.

    Raises:
        BrowserUnavailableError: If called with no native bridge installed.
    """
    value = await send_native_call("storage.list", {})
    keys = value.get("keys", [])
    if not isinstance(keys, list):
        return []
    return [str(key) for key in keys]

put async

put(name: str, content: str) -> None

Write a string value under a storage key, creating or overwriting it.

Parameters:

Name	Type	Description	Default
name	str	The storage key (an IndexedDB key, scoped to the origin).	required
content	str	The string value to store.	required

Raises:

Type	Description
NativeError	If the write fails, e.g. the quota is exceeded (quota_exceeded).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/storage.py

async def put(name: str, content: str) -> None:
    """Write a string value under a storage key, creating or overwriting it.

    Args:
        name: The storage key (an IndexedDB key, scoped to the origin).
        content: The string value to store.

    Raises:
        NativeError: If the write fails, e.g. the quota is exceeded
            (``quota_exceeded``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    await send_native_call("storage.put", {"name": name, "content": content})

remove async

remove(name: str) -> None

Delete the value stored under a key.

Parameters:

Name	Type	Description	Default
name	str	The storage key (an IndexedDB key, scoped to the origin).	required

Raises:

Type	Description
NativeError	If the key does not exist (not_found).
BrowserUnavailableError	If called with no native bridge installed.

Source code in tempestweb/native/storage.py

async def remove(name: str) -> None:
    """Delete the value stored under a key.

    Args:
        name: The storage key (an IndexedDB key, scoped to the origin).

    Raises:
        NativeError: If the key does not exist (``not_found``).
        BrowserUnavailableError: If called with no native bridge installed.
    """
    await send_native_call("storage.remove", {"name": name})