Gateway API

Helpers that translate HTTP requests into Zenoh selectors and map Istos error codes to HTTP status. Used by the embedded aiohttp surface.

Details: HTTP Gateway.

HTTP → Zenoh helpers for the ingress gateway.

Turns an HTTP request into a Zenoh query so @handle / @stream run the usual pipeline (auth, validation, DI, middleware):

POST /robot/move  {"distance": 5}  Authorization: Bearer <token>
    → session.get("robot/move?distance=5", attachment=<token>)
    → @handle("robot/move")
    → JSON (or SSE chunks for ``@stream``)

Parsing / encoding / status mapping live here (no aiohttp). Wire-up is in :mod:istos.app.

HttpRoute dataclass

HTTP path mapped to a Zenoh key. sse=True → stream as SSE.

Source code in src/istos/http/gateway.py

@dataclass
class HttpRoute:
    """HTTP path mapped to a Zenoh key. ``sse=True`` → stream as SSE."""

    method: str
    path: str
    key_expr: str
    timeout_s: float = 5.0
    sse: bool = False

build_selector(key_expr, params)

Combine a key expression with encoded params into a Zenoh selector.

Zenoh separates selector parameters with ; (not &); keys and values are percent-encoded so reserved characters survive.

Source code in src/istos/http/gateway.py

def build_selector(key_expr: str, params: Dict[str, Any]) -> str:
    """Combine a key expression with encoded params into a Zenoh selector.

    Zenoh separates selector parameters with ``;`` (not ``&``); keys and values
    are percent-encoded so reserved characters survive.
    """
    encoded = encode_params(params)
    if not encoded:
        return key_expr
    query = ";".join(f"{quote(str(k))}={quote(str(v))}" for k, v in encoded.items())
    return f"{key_expr}?{query}"

decode_params(raw)

Percent-decode Zenoh selector parameters on the server side.

Clients percent-encode keys/values (so ;, =, spaces survive the selector syntax), but Zenoh does not decode them on receipt — so handlers must. Decodes once, matching the single quote in :func:build_selector and the query client.

Source code in src/istos/http/gateway.py

def decode_params(raw: Dict[str, str]) -> Dict[str, str]:
    """Percent-decode Zenoh selector parameters on the server side.

    Clients percent-encode keys/values (so ``;``, ``=``, spaces survive the
    selector syntax), but Zenoh does **not** decode them on receipt — so handlers
    must. Decodes once, matching the single ``quote`` in :func:`build_selector`
    and the query client.
    """
    return {unquote(k): unquote(v) for k, v in raw.items()}

encode_params(params)

Stringify request params for a Zenoh selector query string.

Scalars become their string form; None is dropped; nested objects/arrays are JSON-encoded (the handler's validator can parse them back).

Source code in src/istos/http/gateway.py

def encode_params(params: Dict[str, Any]) -> Dict[str, str]:
    """Stringify request params for a Zenoh selector query string.

    Scalars become their string form; ``None`` is dropped; nested
    objects/arrays are JSON-encoded (the handler's validator can parse them back).
    """
    out: Dict[str, str] = {}
    for k, v in params.items():
        if v is None:
            continue
        if isinstance(v, bool):
            out[k] = "true" if v else "false"
        elif isinstance(v, (str, int, float)):
            out[k] = str(v)
        else:
            out[k] = json.dumps(v)
    return out

extract_bearer(auth_header)

Pull the token out of an Authorization header.

Accepts Bearer <token> (case-insensitive scheme) or a bare token.

Source code in src/istos/http/gateway.py

def extract_bearer(auth_header: Optional[str]) -> Optional[str]:
    """Pull the token out of an ``Authorization`` header.

    Accepts ``Bearer <token>`` (case-insensitive scheme) or a bare token.
    """
    if not auth_header:
        return None
    parts = auth_header.split(None, 1)
    if len(parts) == 2 and parts[0].lower() == "bearer":
        return parts[1].strip()
    return auth_header.strip()

is_error_payload(parsed)

Whether a decoded reply is an :class:ErrorResponse wire payload.

A handler that raises replies with an envelope rather than sending an exception, and the envelope answers .get() like any other dict, so an unchecked caller reads a failure as data.

query_once, @query, stream_query and open_channel check this themselves. Use it directly for replies you decode yourself, and for multi-reply results, which are passed through unchecked. Pair it with :func:error_from_payload to raise.

Detection prefers the :data:ERROR_MARKER discriminator: present and truthy is an error, present and falsy is a normal result — the escape hatch for a handler whose success value legitimately carries error/code/ message. When the marker is absent (an old responder, or a client in another language), fall back to the legacy rule: all three of error, code and message present.

Source code in src/istos/errors.py

def is_error_payload(parsed: Any) -> bool:
    """Whether a decoded reply is an :class:`ErrorResponse` wire payload.

    A handler that raises replies with an envelope rather than sending an
    exception, and the envelope answers ``.get()`` like any other dict, so an
    unchecked caller reads a failure as data.

    ``query_once``, ``@query``, ``stream_query`` and ``open_channel`` check this
    themselves. Use it directly for replies you decode yourself, and for
    multi-reply results, which are passed through unchecked. Pair it with
    :func:`error_from_payload` to raise.

    Detection prefers the :data:`ERROR_MARKER` discriminator: present and truthy
    is an error, present and falsy is a normal result — the escape hatch for a
    handler whose success value legitimately carries ``error``/``code``/
    ``message``. When the marker is absent (an old responder, or a client in
    another language), fall back to the legacy rule: all three of ``error``,
    ``code`` and ``message`` present.
    """
    if not isinstance(parsed, dict):
        return False
    marker = parsed.get(ERROR_MARKER)
    if marker is not None:
        return bool(marker)
    return all(field in parsed for field in ("error", "code", "message"))

parse_http_spec(spec, prefix, timeout_s=5.0, *, sse=False)

Turn a handler's http= value into an :class:HttpRoute.

• True → POST /<prefix> (GET for sse=True)
• "/custom/path" → POST /custom/path (GET for sse=True)
• "GET /things" → method + path

SSE routes default to GET (what EventSource uses); an explicit method wins.

Source code in src/istos/http/gateway.py

def parse_http_spec(
    spec: Union[bool, str],
    prefix: str,
    timeout_s: float = 5.0,
    *,
    sse: bool = False,
) -> HttpRoute:
    """Turn a handler's ``http=`` value into an :class:`HttpRoute`.

    * ``True``            → ``POST /<prefix>`` (``GET`` for ``sse=True``)
    * ``"/custom/path"``  → ``POST /custom/path`` (``GET`` for ``sse=True``)
    * ``"GET /things"``   → method + path

    SSE routes default to ``GET`` (what ``EventSource`` uses); an explicit method
    wins.
    """
    default_method = "GET" if sse else "POST"
    default_path = "/" + prefix.lstrip("/")
    if spec is True:
        return HttpRoute(default_method, default_path, prefix, timeout_s, sse)
    if isinstance(spec, str):
        parts = spec.split()
        if len(parts) == 1:
            method, path = default_method, parts[0]
        elif len(parts) == 2:
            method, path = parts[0], parts[1]
        else:
            raise ValueError(
                f"Invalid http spec {spec!r}: expected 'METHOD /path', '/path', or True."
            )
        if not path.startswith("/"):
            path = "/" + path
        return HttpRoute(method.upper(), path, prefix, timeout_s, sse)
    raise ValueError(f"Invalid http spec {spec!r}: expected str or True.")

sse_event(data, event=None, *, id=None)

Format one SSE frame. Multi-line data becomes one data: line each; a blank line terminates the frame.

Source code in src/istos/http/gateway.py

def sse_event(data: str, event: Optional[str] = None, *, id: Optional[str] = None) -> str:
    """Format one SSE frame. Multi-line ``data`` becomes one ``data:`` line each;
    a blank line terminates the frame."""
    lines = []
    if id is not None:
        lines.append(f"id: {id}")
    if event is not None:
        lines.append(f"event: {event}")
    for line in data.split("\n"):
        lines.append(f"data: {line}")
    return "\n".join(lines) + "\n\n"

status_for_reply(parsed)

HTTP status for a decoded handler reply (200, or mapped from its error code).

Source code in src/istos/http/gateway.py

def status_for_reply(parsed: Any) -> int:
    """HTTP status for a decoded handler reply (200, or mapped from its error code)."""
    if is_error_payload(parsed):
        return CODE_TO_STATUS.get(parsed["code"], DEFAULT_ERROR_STATUS)
    return 200