Module

templating.streaming

Async stream orchestration for progressive rendering.

When aStream()return value contains awaitables (coroutines) in its context, this module resolves them concurrently using anyio, then feeds the resolved context to kida's synchronousrender_stream()for progressive chunk delivery.

This is Chirp-level orchestration around existing Kida primitives — no changes to Kida's rendering engine required.

Pipeline::

Stream("page.html",
    header=site_header(),        # already resolved (str)
    stats=db.fetch(Stats, ..),   # awaitable (coroutine)
    feed=db.fetch(Event, ..),    # awaitable (coroutine)
)

1. Detect awaitables in context
2. Resolve all awaitables concurrently (anyio.create_task_group)
3. Pass fully-resolved context to kida render_stream()
4. Yield HTML chunks via chunked transfer encoding
3Functions

Functions

resolve_stream_context 1 dict[str, Any] ▼
Resolve any awaitables in a Stream() context concurrently. Values that are cor…
async
async def resolve_stream_context(context: dict[str, Any]) -> dict[str, Any]

Resolve any awaitables in a Stream() context concurrently.

Values that are coroutines or awaitables are resolved in parallel. All other values pass through unchanged.

Returns a new dict with all values fully resolved.

Parameters
Name Type Description
context dict[str, Any]
Returns
dict[str, Any]
render_stream_async 2 AsyncIterator[str] ▼
Render a Stream() with async source resolution. 1. Resolves any awaitable cont…
async
async def render_stream_async(env: Environment, stream: Stream) -> AsyncIterator[str]

Render a Stream() with async source resolution.

  1. Resolves any awaitable context values concurrently
  2. Delegates to kida's synchronous render_stream() for chunk generation
  3. Yields chunks as an async iterator for ASGI consumption

Usage from negotiation.py::

async for chunk in render_stream_async(kida_env, stream_value):
    await send_chunk(chunk)
Parameters
Name Type Description
env Environment
stream Stream
Returns
AsyncIterator[str]
has_async_context 1 bool ▼
Check if a Stream() context contains any awaitables. Used by negotiation.py to…
def has_async_context(context: dict[str, Any]) -> bool

Check if a Stream() context contains any awaitables.

Used by negotiation.py to decide between sync and async rendering paths.

Parameters
Name Type Description
context dict[str, Any]
Returns
bool