`ros_sugar.ui_node.api_utils`#

API helpers built on Starlette, which serves the API and the browser front end

Module Contents#

Classes#

`SameOriginGuard`	Middleware refusing commands another site’s page sends through a browser.
`NoFraming`	Middleware forbidding other sites to frame the pages (clickjacking).
`ApiKeyGuard`	Middleware requiring an API key on the API, or the browser front end’s cookie.

Functions#

`topic_schema`	Field schema for a topic’s ROS message type (empty dict on failure).
`msg_to_jsonable`	JSON-serializable representation of a ROS message (e.g. a service response).
`content_to_jsonable`	JSON-serialize cached output content.
`marker_from_content`	Turn an overlay/path payload into a map op.
`name_param`	Interface name from the URL path, normalized like `Topic` names. Routes use `{name:path}` so namespaced names (ns/topic) stay addressable.
`json_body`	Parse a request’s JSON body.
`reject_websocket`	Refuse a WebSocket for a name the recipe did not declare.
`stream_at_rate`	Accept a websocket and push sampled JSON at the client’s requested rate.
`stream_pushed`	Accept a websocket and push JSON the moment new data arrives (no sampling)
`server_config`	The uvicorn configuration serving `app`, over TLS when a certificate is given.
`is_command`	Whether a request is a command. Any request but GET, HEAD and OPTIONS, and any WebSocket not in `read_streams`.

API#

ros_sugar.ui_node.api_utils.topic_schema(topic) → Dict[str, Any]#: Field schema for a topic’s ROS message type (empty dict on failure).

ros_sugar.ui_node.api_utils.msg_to_jsonable(msg: Any) → Any#: JSON-serializable representation of a ROS message (e.g. a service response).

ros_sugar.ui_node.api_utils.content_to_jsonable(content: Any) → Any#

JSON-serialize cached output content.

Specialized callbacks already return JSON-native content; types without a specialized _get_ui_content yield a raw ROS message, which is converted here.

ros_sugar.ui_node.api_utils.marker_from_content(topic_name: str, type_name: str, content: Any) → Optional[Dict]#

Turn an overlay/path payload into a map op.

Point/Pose/Odometry become {op:"overlay", x, y[, theta]} markers; Path becomes {op:"path", points}. Returns None if there is nothing to render.

ros_sugar.ui_node.api_utils.name_param(conn) → str#: Interface name from the URL path, normalized like Topic names. Routes use {name:path} so namespaced names (ns/topic) stay addressable.

async ros_sugar.ui_node.api_utils.json_body(request) → Any#

Parse a request’s JSON body.

An empty body is an empty request, {}, so body-less calls (e.g. a Trigger service) still work. A body that is not valid JSON is None, which callers reject: treating it as {} would publish a default message.

async ros_sugar.ui_node.api_utils.reject_websocket(websocket, reason: str) → None#

Refuse a WebSocket for a name the recipe did not declare.

Accepted first and then closed with 1008 and a reason. Closing before accepting makes the server answer the handshake with a bare HTTP 403.

async ros_sugar.ui_node.api_utils.stream_at_rate(websocket, default_rate, max_rate, sample) → None#

Accept a websocket and push sampled JSON at the client’s requested rate.

Reads an optional ?rate=<hz> query param (clamped to max_rate). sample() returns (payload, done): payload is sent as JSON when not None; the loop stops when done is True or the client disconnects. The caller must validate the resource before calling this (it accepts the connection).

async ros_sugar.ui_node.api_utils.stream_pushed(websocket, subscribe, unsubscribe, sample) → None#: Accept a websocket and push JSON the moment new data arrives (no sampling)

ros_sugar.ui_node.api_utils.server_config(app: Any, port: int, certificate: Optional[ros_sugar.ui_node.security.Certificate])#: The uvicorn configuration serving app, over TLS when a certificate is given.

ros_sugar.ui_node.api_utils.is_command(scope, read_streams: Sequence[str]) → bool#: Whether a request is a command. Any request but GET, HEAD and OPTIONS, and any WebSocket not in read_streams.

class ros_sugar.ui_node.api_utils.SameOriginGuard(app, open_streams: Sequence[str] = ())#

Middleware refusing commands another site’s page sends through a browser.

A command is as is_command decides, with open_streams as the streams that only send data out. One whose Origin host is not the request’s Host or X-Forwarded-Host is refused with 403, or a WebSocket close 1008. Clients that are not browsers send no Origin and pass.

Parameters:

app – The ASGI app to guard
open_streams – fnmatch patterns of WebSocket paths any site may open, streams that only send data out

class ros_sugar.ui_node.api_utils.NoFraming(app)#: Middleware forbidding other sites to frame the pages (clickjacking).

class ros_sugar.ui_node.api_utils.ApiKeyGuard(app, keys: ros_sugar.ui_node.security.ApiKeys, session_key: Optional[str] = None, prefix: str = '/api/', open_paths: Sequence[str] = (), read_streams: Sequence[str] = (), logger: Any = None)#

Middleware requiring an API key on the API, or the browser front end’s cookie.

A request under prefix, other than open_paths, needs Authorization: Bearer <key> with the command scope for a command (as is_command decides) and the read scope otherwise. It is refused with 401 without a valid key and 403 without the scope, or a WebSocket close 1008.

With a session_key, any other response sets a cookie granting the API to the browser that loaded the page. It is HttpOnly, Secure and SameSite=Strict, so other sites’ pages cannot use it.