App & nodes

App

The blueprint. Collects node definitions from the decorators and resolves the dependency graph.

App

App(title: str = 'Golit App')

Source code in python/golit/app.py

def __init__(self, title: str = "Golit App") -> None:
    self.title = title
    self._defs: dict[str, NodeDef] = {}
    self._order: list[str] = []
    self._widgets: dict[str, Widget] = {}
    self._chat_handlers: dict[str | None, NodeFn] = {}
    self._streams: dict[str, NodeFn] = {}
    self._shared_streams: set[str] = set()
    self._frame_handlers: dict[str, NodeFn] = {}
    self._audio_handlers: dict[str, NodeFn] = {}
    self._pollers: dict[str, tuple[NodeFn, float]] = {}
    self._poll_cache: dict[str, Any] = {}
    self._built = False
    #: Optional page layout tree (see :mod:`golit.layout`); ``None`` stacks
    #: every view under one controls panel.
    self.layout: Any = None

title instance-attribute

title = title

layout instance-attribute

layout: Any = None

compute_ids property

compute_ids: set[str]

node_defs property

node_defs: dict[str, NodeDef]

widgets property

widgets: dict[str, Widget]

chat_handlers property

chat_handlers: dict[str | None, NodeFn]

Registered chat message handlers, keyed by channel (None = all).

streams property

streams: dict[str, NodeFn]

Registered video frame producers, keyed by stream name.

shared_streams property

shared_streams: set[str]

Names of streams registered with shared=True (one producer, fanned out).

frame_handlers property

frame_handlers: dict[str, NodeFn]

Registered browser-camera frame processors, keyed by camera name.

audio_handlers property

audio_handlers: dict[str, NodeFn]

Registered browser-mic clip handlers, keyed by recorder name.

pollers property

pollers: dict[str, tuple[NodeFn, float]]

Registered live polled sources, keyed by node name → (fetch fn, interval seconds).

poll_cache property

poll_cache: dict[str, Any]

The latest value of each polled source; the background poller writes it, the source node reads it (per worker process).

source

source(fn: NodeFn) -> NodeFn

Register a source node (brings data in — read a file, query a DB, return a sample frame). May depend on inputs. Returns fn unchanged.

Source code in python/golit/app.py

def source(self, fn: NodeFn) -> NodeFn:
    """Register a **source** node (brings data in — read a file, query a DB,
    return a sample frame). May depend on inputs. Returns ``fn`` unchanged."""
    return self._register(fn, NodeKind.SOURCE)

reactive

reactive(fn: NodeFn) -> NodeFn

Register a reactive node — a pure transform over its upstream nodes and inputs. Re-runs only when one of those changes. Returns fn unchanged.

Source code in python/golit/app.py

def reactive(self, fn: NodeFn) -> NodeFn:
    """Register a **reactive** node — a pure transform over its upstream nodes
    and inputs. Re-runs only when one of those changes. Returns ``fn`` unchanged."""
    return self._register(fn, NodeKind.REACTIVE)

view

view(fn: NodeFn) -> NodeFn

Register a view node — a renderable leaf whose return value Golit turns into a UI fragment. Re-renders only when an input changes. Returns fn unchanged.

Source code in python/golit/app.py

def view(self, fn: NodeFn) -> NodeFn:
    """Register a **view** node — a renderable leaf whose return value Golit
    turns into a UI fragment. Re-renders only when an input changes. Returns
    ``fn`` unchanged."""
    return self._register(fn, NodeKind.VIEW)

on_message

on_message(channel: Any = None) -> Any

Register a handler for incoming chat messages on channel (or all channels when None). Without a handler a channel simply relays every message to the room; with one, the handler owns the message and responds via the :class:~golit.server.chat.MessageContext it's given::

@app.on_message("room")
async def handle(msg, ctx):
    await ctx.broadcast(msg.text, author=msg.author)   # relay
    if msg.text.startswith("/bot"):
        await ctx.reply("beep boop", author="Bot")     # to sender only

Usable bare (@app.on_message) or with a channel (@app.on_message("room")). The handler may be sync or async.

Source code in python/golit/app.py

def on_message(self, channel: Any = None) -> Any:
    """Register a handler for incoming chat messages on ``channel`` (or all
    channels when ``None``). Without a handler a channel simply relays every
    message to the room; with one, the handler owns the message and responds
    via the :class:`~golit.server.chat.MessageContext` it's given::

        @app.on_message("room")
        async def handle(msg, ctx):
            await ctx.broadcast(msg.text, author=msg.author)   # relay
            if msg.text.startswith("/bot"):
                await ctx.reply("beep boop", author="Bot")     # to sender only

    Usable bare (``@app.on_message``) or with a channel (``@app.on_message("room")``).
    The handler may be sync or async."""
    if callable(channel):  # used bare: @app.on_message
        self._chat_handlers[None] = channel
        return channel

    def deco(fn: NodeFn) -> NodeFn:
        self._chat_handlers[channel] = fn
        return fn

    return deco

stream

stream(
    name: str, *, shared: bool = False
) -> Callable[[NodeFn], NodeFn]

Register a video frame producer named name, shown by :func:golit.ui.webcam. The decorated function returns an iterator of frames — JPEG bytes (e.g. from cv2.imencode) or (H, W, 3) uint8 RGB arrays (encoded for you) — and each request starts a fresh stream that Golit pushes as an MJPEG (multipart/x-mixed-replace) response off the event loop::

@app.stream("camera")
def camera():
    cap = cv2.VideoCapture(0)
    try:
        while True:
            ok, frame = cap.read()
            if not ok:
                break
            # ... run your detector and draw on `frame` ...
            yield cv2.imencode(".jpg", frame)[1].tobytes()
    finally:
        cap.release()

Sync or async (async def + yield) producers both work; sync frames are pulled in a worker thread so a blocking camera read or model never stalls the loop.

With shared=True the producer runs once no matter how many viewers connect: a single background pull keeps the latest frame and fans it out to every <img>, so N viewers of one camera don't open N devices. The producer starts on the first viewer and its finally runs when the last leaves. Leave it False (the default) for synthetic feeds or a genuinely per-viewer source.

Source code in python/golit/app.py

def stream(self, name: str, *, shared: bool = False) -> Callable[[NodeFn], NodeFn]:
    """Register a video **frame producer** named ``name``, shown by
    :func:`golit.ui.webcam`. The decorated function returns an iterator of frames —
    JPEG ``bytes`` (e.g. from ``cv2.imencode``) or ``(H, W, 3)`` uint8 RGB arrays
    (encoded for you) — and each request starts a fresh stream that Golit pushes as an
    MJPEG (``multipart/x-mixed-replace``) response off the event loop::

        @app.stream("camera")
        def camera():
            cap = cv2.VideoCapture(0)
            try:
                while True:
                    ok, frame = cap.read()
                    if not ok:
                        break
                    # ... run your detector and draw on `frame` ...
                    yield cv2.imencode(".jpg", frame)[1].tobytes()
            finally:
                cap.release()

    Sync or async (``async def`` + ``yield``) producers both work; sync frames are
    pulled in a worker thread so a blocking camera read or model never stalls the loop.

    With ``shared=True`` the producer runs **once** no matter how many viewers connect:
    a single background pull keeps the latest frame and fans it out to every ``<img>``,
    so N viewers of one camera don't open N devices. The producer starts on the first
    viewer and its ``finally`` runs when the last leaves. Leave it ``False`` (the default)
    for synthetic feeds or a genuinely per-viewer source."""

    def deco(fn: NodeFn) -> NodeFn:
        self._streams[name] = fn
        if shared:
            self._shared_streams.add(name)
        return fn

    return deco

on_frame

on_frame(name: str) -> Callable[[NodeFn], NodeFn]

Register a per-frame processor for the browser-camera view named name, shown by :func:golit.ui.camera. Each frame the visitor's webcam captures is sent up (over a WebSocket), decoded to an (H, W, 3) uint8 RGB array, passed to the handler, and the value it returns — an RGB array or JPEG bytes — is sent back and displayed::

@app.on_frame("detector")
def detect(frame):           # frame: (H, W, 3) uint8 RGB
    # ... run your model and draw on a copy of `frame` ...
    return frame             # annotated RGB array (or JPEG bytes)

Sync or async (async def) handlers both work; sync ones (and every JPEG decode/encode) run in a worker thread so a heavy model never stalls the loop. One frame is in flight at a time — the client waits for each result — so a slow handler simply lowers the frame rate instead of building a backlog. The mirror of :meth:stream, which produces frames on the server; here the browser is the camera and the server transforms.

Source code in python/golit/app.py

def on_frame(self, name: str) -> Callable[[NodeFn], NodeFn]:
    """Register a per-frame **processor** for the browser-camera view named ``name``,
    shown by :func:`golit.ui.camera`. Each frame the visitor's webcam captures is sent up
    (over a WebSocket), decoded to an ``(H, W, 3)`` uint8 RGB array, passed to the handler,
    and the value it returns — an RGB array or JPEG ``bytes`` — is sent back and displayed::

        @app.on_frame("detector")
        def detect(frame):           # frame: (H, W, 3) uint8 RGB
            # ... run your model and draw on a copy of `frame` ...
            return frame             # annotated RGB array (or JPEG bytes)

    Sync or async (``async def``) handlers both work; sync ones (and every JPEG
    decode/encode) run in a worker thread so a heavy model never stalls the loop. One frame
    is in flight at a time — the client waits for each result — so a slow handler simply
    lowers the frame rate instead of building a backlog. The mirror of :meth:`stream`, which
    produces frames on the server; here the browser is the camera and the server transforms."""

    def deco(fn: NodeFn) -> NodeFn:
        self._frame_handlers[name] = fn
        return fn

    return deco

on_audio

on_audio(name: str) -> Callable[[NodeFn], NodeFn]

Register a handler for clips recorded by the browser-mic view named name, shown by :func:golit.ui.recorder. When the visitor stops recording, the clip is uploaded (over a WebSocket) as 16-bit PCM WAV bytes and passed to the handler. Whatever it returns is shown back in the recorder::

import wave, io

@app.on_audio("note")
def transcribe(wav: bytes):              # 16-bit PCM WAV bytes
    with wave.open(io.BytesIO(wav)) as w:  # stdlib — no ffmpeg needed
        seconds = w.getnframes() / w.getframerate()
    return f"<p>{seconds:.1f}s recorded</p>"   # HTML/renderable, shown in place

A renderable return (string, DataFrame, a :mod:golit.ui component, …) is rendered and displayed; returning bytes instead sends audio back for playback (an echo, a TTS reply). Sync or async handlers both work — sync ones run in a worker thread so a heavy transcribe never stalls the loop. The audio mirror of :meth:on_frame.

Source code in python/golit/app.py

def on_audio(self, name: str) -> Callable[[NodeFn], NodeFn]:
    """Register a handler for clips recorded by the browser-mic view named ``name``,
    shown by :func:`golit.ui.recorder`. When the visitor stops recording, the clip is
    uploaded (over a WebSocket) as 16-bit PCM **WAV** ``bytes`` and passed to the handler.
    Whatever it returns is shown back in the recorder::

        import wave, io

        @app.on_audio("note")
        def transcribe(wav: bytes):              # 16-bit PCM WAV bytes
            with wave.open(io.BytesIO(wav)) as w:  # stdlib — no ffmpeg needed
                seconds = w.getnframes() / w.getframerate()
            return f"<p>{seconds:.1f}s recorded</p>"   # HTML/renderable, shown in place

    A renderable return (string, DataFrame, a :mod:`golit.ui` component, …) is rendered and
    displayed; returning **`bytes`** instead sends audio back for playback (an echo, a TTS
    reply). Sync or async handlers both work — sync ones run in a worker thread so a heavy
    transcribe never stalls the loop. The audio mirror of :meth:`on_frame`."""

    def deco(fn: NodeFn) -> NodeFn:
        self._audio_handlers[name] = fn
        return fn

    return deco

poll

poll(
    name: str, *, interval: float = 2.0
) -> Callable[[NodeFn], NodeFn]

Register a live polled source named name — external data that changes on its own (a Google Sheet, an API, a file). The decorated function fetches the data; Golit runs it every interval seconds in the background, and when the result's content hash changes it updates the source and pushes the dependent views to every connected browser over SSE — the same server-side-invalidation path streaming sources use, so you write a plain @app.view and it just updates::

@app.poll("sheet", interval=3)
async def sheet() -> pl.DataFrame:
    return await fetch_csv(SHEET_URL)     # whatever fetch you like

@app.view
def table(sheet) -> str:                  # depends on the polled source by name
    return ui.table(sheet) if sheet is not None else ui.spinner(label="Loading…")

The fetch may be sync or async; sync fetches run in a worker thread so a blocking request never stalls the loop. Until the first fetch lands the source is None — have views handle that. One poller runs per worker process. A fetch that raises is logged and retried on the next tick (the last good value stays).

Source code in python/golit/app.py

def poll(self, name: str, *, interval: float = 2.0) -> Callable[[NodeFn], NodeFn]:
    """Register a **live polled source** named ``name`` — external data that changes on
    its own (a Google Sheet, an API, a file). The decorated function *fetches* the data;
    Golit runs it every ``interval`` seconds in the background, and when the result's
    content hash changes it updates the source and pushes the dependent views to every
    connected browser over SSE — the same server-side-invalidation path streaming sources
    use, so you write a plain ``@app.view`` and it just updates::

        @app.poll("sheet", interval=3)
        async def sheet() -> pl.DataFrame:
            return await fetch_csv(SHEET_URL)     # whatever fetch you like

        @app.view
        def table(sheet) -> str:                  # depends on the polled source by name
            return ui.table(sheet) if sheet is not None else ui.spinner(label="Loading…")

    The fetch may be sync or ``async``; sync fetches run in a worker thread so a blocking
    request never stalls the loop. Until the first fetch lands the source is ``None`` —
    have views handle that. One poller runs per worker process. A fetch that raises is
    logged and retried on the next tick (the last good value stays)."""
    if interval <= 0:
        raise ValueError("poll interval must be positive")

    def deco(fn: NodeFn) -> NodeFn:
        if name in self._defs:
            raise ValueError(f"duplicate node {name!r}")
        self._pollers[name] = (fn, float(interval))

        def reader() -> Any:
            return self._poll_cache.get(name)

        reader.__name__ = name
        self._register(reader, NodeKind.SOURCE)
        return fn

    return deco

build

build() -> None

Resolve every parameter to an input/dependency/constant. Raises if a parameter is neither a widget, a known node, nor a defaulted constant.

Source code in python/golit/app.py

def build(self) -> None:
    """Resolve every parameter to an input/dependency/constant. Raises if a
    parameter is neither a widget, a known node, nor a defaulted constant."""
    for node_id, ndef in self._defs.items():
        deps: list[str] = []
        for p in ndef.params:
            if p.widget is not None:
                deps.append(p.name)  # edge to the input node
            elif p.name in self._defs:
                deps.append(p.name)  # edge to another compute node
            elif p.has_default:
                continue  # constant kwarg
            else:
                raise ValueError(
                    f"node {node_id!r}: parameter {p.name!r} is not a known node "
                    f"or widget input (and has no default)"
                )
        ndef.deps = deps
        if ndef.kind is NodeKind.VIEW:
            ndef.target = node_id
    if self.layout is not None:
        from .layout import validate_layout

        validate_layout(self.layout, self)
    self._built = True

new_graph

new_graph() -> Graph

Build a fresh kernel graph for a session (topology only; state is reset per session).

Source code in python/golit/app.py

def new_graph(self) -> Graph:
    """Build a fresh kernel graph for a session (topology only; state is
    reset per session)."""
    if not self._built:
        self.build()
    graph = Graph()
    for input_id in self._widgets:
        graph.add_node(input_id, NodeKind.INPUT.value)
    for node_id, ndef in self._defs.items():
        graph.add_node(node_id, ndef.kind.value)
    for node_id, ndef in self._defs.items():
        if ndef.deps:
            graph.set_deps(node_id, ndef.deps)
    graph.build()
    return graph

node_def

node_def(node_id: str) -> NodeDef

Source code in python/golit/app.py

def node_def(self, node_id: str) -> NodeDef:
    return self._defs[node_id]

widget_for

widget_for(input_id: str) -> Widget | None

Source code in python/golit/app.py

def widget_for(self, input_id: str) -> Widget | None:
    return self._widgets.get(input_id)

input_default

input_default(input_id: str) -> Any

Source code in python/golit/app.py

def input_default(self, input_id: str) -> Any:
    return self._widgets[input_id].default

Node kinds & definitions

NodeKind

Bases: StrEnum

INPUT class-attribute instance-attribute

INPUT = 'input'

SOURCE class-attribute instance-attribute

SOURCE = 'source'

REACTIVE class-attribute instance-attribute

REACTIVE = 'reactive'

VIEW class-attribute instance-attribute

VIEW = 'view'

NodeDef dataclass

NodeDef(
    id: str,
    kind: NodeKind,
    fn: Callable[..., Any],
    params: list[Param],
    deps: list[str] = list(),
    target: str | None = None,
)

The blueprint for a single node (resolved deps filled in by App.build).

id instance-attribute

id: str

kind instance-attribute

kind: NodeKind

fn instance-attribute

fn: Callable[..., Any]

params instance-attribute

params: list[Param]

deps class-attribute instance-attribute

deps: list[str] = field(default_factory=list)

target class-attribute instance-attribute

target: str | None = None

Param dataclass

Param(
    name: str,
    widget: Widget | None,
    has_default: bool,
    default: Any,
)

One parameter of a node function.

name instance-attribute

name: str

widget instance-attribute

widget: Widget | None

has_default instance-attribute

has_default: bool

default instance-attribute

default: Any

inspect_params

inspect_params(fn: Callable[..., Any]) -> list[Param]

Extract :class:Param metadata from a function signature.

Source code in python/golit/nodes.py

def inspect_params(fn: Callable[..., Any]) -> list[Param]:
    """Extract :class:`Param` metadata from a function signature."""
    params: list[Param] = []
    for name, p in inspect.signature(fn).parameters.items():
        if p.kind in (p.VAR_POSITIONAL, p.VAR_KEYWORD):
            continue
        default = p.default
        if isinstance(default, Widget):
            params.append(Param(name, widget=default, has_default=True, default=None))
        elif default is inspect.Parameter.empty:
            params.append(Param(name, widget=None, has_default=False, default=None))
        else:
            params.append(Param(name, widget=None, has_default=True, default=default))
    return params

Session

The per-client scheduler driver — a fresh kernel graph + value registry built from the shared App blueprint.

Session

Session(
    app: App, *, view_renderer: ViewRenderer | None = None
)

Per-session reactive state: a fresh kernel graph + value registry built from the shared :class:App blueprint.

Source code in python/golit/engine.py

def __init__(self, app: App, *, view_renderer: ViewRenderer | None = None) -> None:
    self.app = app
    self.graph = app.new_graph()
    self.registry = Registry()
    self._render = view_renderer or _default_renderer

app instance-attribute

app = app

graph instance-attribute

graph = new_graph()

registry instance-attribute

registry = Registry()

initial_render

initial_render() -> dict[str, str]

Compute the whole graph once; return every view fragment.

Source code in python/golit/engine.py

def initial_render(self) -> dict[str, str]:
    """Compute the whole graph once; return every view fragment."""
    return self._run(self.graph.topo_order(), force=True)

update

update(input_id: str, raw_value: Any) -> dict[str, str]

Commit an input change and return only the view fragments that changed.

Source code in python/golit/engine.py

def update(self, input_id: str, raw_value: Any) -> dict[str, str]:
    """Commit an input change and return only the view fragments that changed."""
    widget = self.app.widget_for(input_id)
    if widget is None:
        raise KeyError(f"unknown input {input_id!r}")
    value = widget.coerce(raw_value)
    self.registry.set(input_id, value)
    # Push the new content hash to the kernel so downstream signatures shift
    # (or, on a revert to a prior value, stay put for a clean memo hit).
    self.graph.commit_input(input_id, hash_value(value))
    return self._run(self.graph.dirty_subgraph([input_id]))

refresh

refresh(node_id: str) -> dict[str, str]

Force-recompute a node (e.g. a streaming source or a shared node) and everything downstream; return the view fragments that changed. This is the server-initiated path that feeds the SSE push channel.

Source code in python/golit/engine.py

def refresh(self, node_id: str) -> dict[str, str]:
    """Force-recompute a node (e.g. a streaming source or a shared node) and
    everything downstream; return the view fragments that changed. This is the
    server-initiated path that feeds the SSE push channel."""
    return self._run(self.graph.dirty_subgraph([node_id]), force_ids={node_id})

control_html

control_html(input_id: str) -> str

Render an input's HTML control at its current value.

Source code in python/golit/engine.py

def control_html(self, input_id: str) -> str:
    """Render an input's HTML control at its current value."""
    widget = self.app.widget_for(input_id)
    assert widget is not None
    return widget.render(self.registry.get_or(input_id, widget.default))

fragment

fragment(node_id: str) -> str | None

Source code in python/golit/engine.py

def fragment(self, node_id: str) -> str | None:
    return self.registry.fragment(node_id)