Async stream orchestration for progressive rendering.
EveryStream()return value is rendered through this module. When the
context contains awaitables (coroutines) they are resolved concurrently using
anyio first; the resolved context (sync contexts pass straight through) is then
fed to kida's synchronousrender_stream(), driven on a worker thread so the
event loop is never blocked by CPU-bound chunk compilation — for either an
all-sync or a mixed context (issue #179).
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. Drive kida's lazy sync render_stream() on a worker thread, bridging
each chunk back to the loop through a bounded queue
4. Yield HTML chunks via chunked transfer encoding
Thread + bounded-queue bridge (issue #179):
kida'srender_stream()is a CPU-bound synchronous generator. Iterating
it inline on the event loop blocks every concurrent request for the duration
of each chunk's compilation. It also cannot simply be wrapped in
anyio.to_thread.run_sync— that runs a callable to completion, which
would buffer the whole render and defeat progressive flush.
Instead a dedicated worker thread drives the generator chunk-by-chunk and
hands each chunk to the loop through a bounded ``queue.Queue``
(``_STREAM_CHUNK_BUFFER`` rendered chunks in flight, plus one reserved slot
for the terminal sentinel). The loop pulls each chunk via
``anyio.to_thread.run_sync(queue.get)``, so that ``await`` lets concurrent
tasks make progress while the worker computes the next chunk, and the
bounded queue applies back-pressure (the worker blocks in ``put`` when the
loop is slow), keeping memory bounded. The kida template and its generator
are *both created and driven on the same worker thread* — never created on
the loop and iterated on the worker — to respect kida's single-thread
renderer contract. No anyio task group or portal wraps the ``yield``, so a
consumer ``aclose()`` (``GeneratorExit``) unwinds cleanly rather than being
wrapped into a noisy ``ExceptionGroup``.
Cancellation contract: on client disconnect / consumer ``aclose()`` the
shielded ``finally`` block sets a stop event and drains the queue so the
worker's blocked ``put`` unwinds, then joins the worker thread to
completion. The thread cannot leak past the async generator's ``aclose()``.
A mid-stream render error is captured on the worker thread and re-raised on
the loop after the worker exits, so ``sender.py``'s mid-stream error path
still fires.
Shell-first streaming (kida 0.2.3+):
Use{% flush %}in templates to emit a streaming boundary. Place it
after header/nav so the client receives the shell before main content::
<html><head>...</head><body>
<header>...</header><nav>...</nav>
{% flush %}
<main>{% for item in items %}...{% end %}</main>
</body></html>
templating.streaming
| Name | Type | Default | Description |
|---|---|---|---|
type
|
|
— | |
qualified_name
|
|
— | |
element_type
|
|
— | |
description
|
|
— | |
source_file
|
|
— | |
line_number
|
|
— | |
is_autodoc
|
|
— | |
autodoc_element
|
|
— | |
_autodoc_template
|
|
— | |
_autodoc_url_path
|
|
— | |
_autodoc_page_type
|
|
— | |
title
|
|
— | |
doc_content_hash
|
|
— |
Symbols on this page
Resolve any awaitables in a Stream() context concurrently.
Values that are coroutines or awaitables are resolved in parallel. All other values pass through unchanged.…
Render a Stream() with async source resolution, off the event loop.
- Resolves any awaitable context values concurrently
- Drives kida's synchronous
render_stream()on a dedicated…
Check if a Stream() context contains any awaitables.
Used by negotiation.py to decide between sync and async rendering paths.
resolve_stream_context
function
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 | Default | Description |
|---|---|---|---|
context
|
dict[str, Any]
|
— |
_StreamSentinel
class
render_stream_async
function
async
async def render_stream_async(env: Environment, stream: Stream) -> AsyncIterator[str]Render a Stream() with async source resolution, off the event loop.
- Resolves any awaitable context values concurrently
- Drives kida's synchronous
render_stream()on a dedicated worker thread, bridging each chunk back to the loop through a bounded queue so the loop is never blocked by CPU-bound chunk compilation - Yields chunks as an async iterator for ASGI consumption while preserving progressive flush and chunk order
See the module docstring for the threading + cancellation contract.
Usage from negotiation.py::
async for chunk in render_stream_async(kida_env, stream_value):
await send_chunk(chunk)
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
env
|
Environment
|
— | |
stream
|
Stream
|
— |
has_async_context
function
def has_async_context(context: dict[str, Any]) -> boolCheck if a Stream() context contains any awaitables.
Used by negotiation.py to decide between sync and async rendering paths.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
context
|
dict[str, Any]
|
— |
View source · /home/runner/work/chirp/chirp/site/../src/chirp/templating/streaming.py:1