Module

pages.reactive

Structured reactive templates — automatic SSE push of changed blocks.

Uses kida's static dependency analysis (DependencyWalker) to know which template blocks depend on which context paths. When a store mutation changes data, affected blocks are re-rendered via render_block()and pushed over SSE.

Components:

  • ChangeEvent: Emitted by stores after mutations.
  • ReactiveBus: Broadcast channel for change events (per-scope).
  • DependencyIndex: Maps context paths to block references.
  • reactive_stream(): SSE endpoint that auto-pushes affected blocks.

Example::

# In the store
bus = ReactiveBus()

def apply_edit(self, edit: Edit) -> Document | None:
    updated = self._apply_edit_inner(edit)
    if updated:
        bus.emit_sync(ChangeEvent(
            scope=edit.doc_id,
            changed_paths=frozenset({"doc.content", "doc.version"}),
        ))
    return updated

# In the route
@app.route("/doc/{doc_id}/live")
def live(doc_id: str) -> EventStream:
    return reactive_stream(
        bus, scope=doc_id,
        index=dep_index,
        kida_env=env,
        context_builder=lambda: build_context(doc_id),
    )
5Classes2Functions

Classes

ChangeEvent 3 ▼
Emitted by a store after a data mutation.

Emitted by a store after a data mutation.

Attributes

Name Type Description
scope str

Scope identifier (e.g., a document ID) so subscribers only receive events for their scope.
changed_paths frozenset[str]

The set of context paths that changed (e.g.,{"doc.content", "doc.version"}).
origin str | None

Opaque identifier for who caused this change (e.g., user ID, session ID). Used byreactive_stream to skip events originating from the same connection. Nonemeans system-initiated — always delivered.
ReactiveBus 5 ▼
Broadcast channel for data change events. Thread-safe. Each call to ``subscribe(scope)`` returns …

Broadcast channel for data change events.

Thread-safe. Each call tosubscribe(scope)returns an async iterator that yieldsChangeEvents for that scope. When emit()is called, the event is placed into every matching subscriber's queue.

Modeled on chirp'sToolEventBusbut scoped per-key.

Methods

emit_sync 1 ▼
Broadcast a change event synchronously (from any thread). Uses ``put_nowait`` …
def emit_sync(self, event: ChangeEvent) -> None

Broadcast a change event synchronously (from any thread).

Usesput_nowaitso it never blocks. Drops the event for a subscriber if its queue is full (back-pressure).

Parameters
Name Type Description
event —
emit 1 ▼
Broadcast a change event (async version).
async
async def emit(self, event: ChangeEvent) -> None
Parameters
Name Type Description
event —
subscribe 1 AsyncIterator[ChangeEven… ▼
Subscribe to change events for a specific scope. Yields ``ChangeEvent`` object…
async
async def subscribe(self, scope: str) -> AsyncIterator[ChangeEvent]

Subscribe to change events for a specific scope.

YieldsChangeEventobjects as they are emitted. The subscription is automatically cleaned up when the iterator exits (client disconnects).

Parameters
Name Type Description
scope —
Returns
AsyncIterator[ChangeEvent]
close 1 ▼
Signal subscribers to stop. If *scope* is given, only close that scope's subsc…
def close(self, scope: str | None = None) -> None

Signal subscribers to stop.

If scope is given, only close that scope's subscribers. Otherwise close all.

Parameters
Name Type Description
scope — Default:None
Internal Methods 1 ▼
__init__ 0 ▼
def __init__(self) -> None
BlockRef 4 ▼
Reference to a renderable block within a template.

Reference to a renderable block within a template.

Attributes

Name Type Description
template_name str

Kida template name.
block_name str

Block name within the template.
dom_id str | None

DOM element ID to target for OOB swap. Defaults to the block name.

Methods

target_id 0 str ▼
DOM element ID for OOB targeting.
property
def target_id(self) -> str
Returns
str
_SSESwapElement 3 ▼
Parsed info about an HTML element with ``sse-swap``.

Parsed info about an HTML element withsse-swap.

Attributes

Name Type Description
swap_event str —
dom_id str | None —
inner_block str | None —
DependencyIndex 4 ▼
Maps context paths to the template blocks that depend on them. Built at app startup from kida's ``…

Maps context paths to the template blocks that depend on them.

Built at app startup from kida'sBlockMetadata.depends_onsets. Thread-safe after construction (read-only).

Example::

index = DependencyIndex()
index.register_template(env, "doc/{doc_id}/_layout.html")
affected = index.affected_blocks({"doc.version"})
# -> [BlockRef("doc/{doc_id}/_layout.html", "toolbar")]

Methods

register_template 4 ▼
Register a template's blocks into the dependency index. Uses kida's static ana…
def register_template(self, env: Environment, template_name: str, *, block_names: list[str] | None = None, dom_id_map: dict[str, str] | None = None) -> None

Register a template's blocks into the dependency index.

Uses kida's static analysis to extract dependencies.

Parameters
Name Type Description
env —

Kida environment.
template_name —

Template to analyze.
block_names —

If given, only register these blocks. Otherwise all blocks are registered.

Default:None
dom_id_map —

Mapping of block_name -> DOM element ID. If a block isn't in this map, block_name is used.

Default:None
register_from_sse_swaps 4 int ▼
Auto-register blocks that have matching ``sse-swap`` elements. Scans the raw t…
def register_from_sse_swaps(self, env: Environment, template_name: str, template_source: str, *, exclude_blocks: set[str] | None = None) -> int

Auto-register blocks that have matchingsse-swapelements.

Scans the raw template source for elements with bothsse-swap andid attributes, finds the {% block %}inside each, and registers only those blocks — with the correctdom_id mapping.

Blocks listed in exclude_blocks are skipped (e.g., client-managedcontenteditableblocks that should never be re-rendered via SSE).

Returns the number of blocks auto-registered.

Example::

index = DependencyIndex()
source = env.loader.get_source(env, "page.html")[0]
n = index.register_from_sse_swaps(env, "page.html", source)
# Registers only blocks inside sse-swap elements
Parameters
Name Type Description
env —
template_name —
template_source —
exclude_blocks — Default:None
Returns
int
affected_blocks 1 list[BlockRef] ▼
Find all blocks affected by a set of changed context paths. Also checks parent…
def affected_blocks(self, changed_paths: frozenset[str]) -> list[BlockRef]

Find all blocks affected by a set of changed context paths.

Also checks parent paths: if"doc.version"changed and a block depends on"doc", it's considered affected.

Parameters
Name Type Description
changed_paths —
Returns
list[BlockRef] List of unique ``BlockRef`` objects.
Internal Methods 1 ▼
__init__ 0 ▼
def __init__(self) -> None

Functions

_extract_sse_swap_elements 1 list[_SSESwapElement] ▼
Extract elements with ``sse-swap`` and find their associated blocks. For each …
def _extract_sse_swap_elements(source: str) -> list[_SSESwapElement]

Extract elements withsse-swapand find their associated blocks.

For each<tag ... sse-swap="event" ...>found, captures the swap event name, elementid, and the associated {% block NAME %}.

Handles two common patterns:

  1. Block inside element (block is a child)::

    <span id="status" sse-swap="status">
    {% block toolbar_status %}v{{ doc.version }}{% endblock %}
</span>

  2. Block wraps element (block is the parent)::

    {% block toolbar_status %}
<span id="status" sse-swap="status">v{{ doc.version }}</span>
{% endblock %}
Parameters
Name Type Description
source str
Returns
list[_SSESwapElement]
reactive_stream 6 EventStream ▼
Create an SSE EventStream that auto-pushes re-rendered blocks. Subscribes to t…
def reactive_stream(bus: ReactiveBus, *, scope: str, index: DependencyIndex, context_builder: Callable[[], dict[str, Any] | Awaitable[dict[str, Any]]], origin: str | None = None, kida_env: Any = None) -> EventStream

Create an SSE EventStream that auto-pushes re-rendered blocks.

Subscribes to theReactiveBusfor the given scope. When a ChangeEventarrives, looks up affected blocks in the DependencyIndex and yields them as Fragmentobjects. The chirp SSE layer handles rendering via the app's kida env.

Parameters
Name Type Description
bus ReactiveBus

The reactive event bus to subscribe to.
scope str

Scope key (e.g., document ID).
index DependencyIndex

Dependency index mapping paths to blocks.
context_builder Callable[[], dict[str, Any] | Awaitable[dict[str, Any]]]

Callable that returns the current context dict (called after each change to get fresh data).
origin str | None

Identity of this connection (e.g., user/session ID). Events whoseorigin matches are skipped — the client that caused the change doesn't need to be notified of it. Nonedisables origin filtering.

Default:None
kida_env Any

Deprecated — rendering is handled by the SSE response layer. Accepted for backwards compatibility.

Default:None
Returns
EventStream