From e71af1d1aa1790afff5e2a9b8e82326e4c66d8b3 Mon Sep 17 00:00:00 2001 From: philippe Date: Thu, 9 Jul 2026 14:12:15 -0400 Subject: [PATCH 1/3] Add streaming callbacks: @callback(..., stream=True) A callback registered with stream=True is a generator (or async generator) whose yields are pushed to the browser as they are produced. Each yielded value has the same shape as a regular return value and replaces the outputs; yielding dash.Patch gives incremental updates (e.g. LLM token streaming). The last yield is the final value. Transport follows the callback's normal selection: over the WebSocket callback transport, frames ride the open connection as callback_response messages with stream: true; otherwise the HTTP response streams NDJSON (application/x-ndjson), one frame per line with a terminal {"done": true} frame. Works on Flask, Quart, and FastAPI; sync generators warn at registration since they occupy a server worker for the whole stream. - dash/_streaming.py: StreamedCallbackResponse marker, context-safe iteration helpers (callback context travels in a contextvars snapshot so set_props/ctx work after dispatch returns), NDJSON serialization, and a thread bridge for async generators on Flask. - dash/_callback.py: stream wrappers building one frame per yield via _prepare_response; per-yield no_update/PreventUpdate handling, on_error support, error frames; registration validation (mutually exclusive with background/clientside/mcp_enabled/api_endpoint). - backends: streaming response branches in each serve_callback; ws.py frame emitter and stream consumers for both the event-loop and threadpool dispatch paths. - renderer: applyStreamFrame applies frames on arrival through the sideUpdate path (Patch applies exactly once); NDJSON reader in handleServerside; stream-aware callback_response handling keeps the request pending until the terminal frame; loading states span the whole stream. --- .ai/ARCHITECTURE.md | 93 ++++++ CHANGELOG.md | 5 + dash/_callback.py | 254 +++++++++++++++- dash/_streaming.py | 189 ++++++++++++ dash/backends/_fastapi.py | 23 +- dash/backends/_flask.py | 37 +++ dash/backends/_quart.py | 21 ++ dash/backends/ws.py | 115 +++++++ dash/dash-renderer/src/actions/callbacks.ts | 154 +++++++++- dash/dash-renderer/src/types/callbacks.ts | 4 + dash/dash-renderer/src/utils/workerClient.ts | 31 +- dash/exceptions.py | 4 + dash/types.py | 3 + .../callbacks/test_stream_callbacks.py | 217 ++++++++++++++ tests/unit/test_stream_callbacks.py | 283 ++++++++++++++++++ tests/websocket/test_ws_stream.py | 163 ++++++++++ 16 files changed, 1590 insertions(+), 6 deletions(-) create mode 100644 dash/_streaming.py create mode 100644 tests/integration/callbacks/test_stream_callbacks.py create mode 100644 tests/unit/test_stream_callbacks.py create mode 100644 tests/websocket/test_ws_stream.py diff --git a/.ai/ARCHITECTURE.md b/.ai/ARCHITECTURE.md index 6566ef8085..71172846a8 100644 --- a/.ai/ARCHITECTURE.md +++ b/.ai/ARCHITECTURE.md @@ -1053,6 +1053,99 @@ async def validate_message(websocket, message): - `@plotly/dash-websocket-worker/src/worker.ts` - SharedWorker entry point - `dash/backends/_fastapi.py` - Server-side WebSocket handler +## Streaming Callbacks + +`@callback(..., stream=True)` marks a generator (or async generator) callback +whose yields are pushed to the browser as they are produced — for LLM token +streaming, progress feeds, and long computations. + +```python +import asyncio +from dash import callback, Output, Input, Patch + +@callback( + Output('log', 'children'), + Input('btn', 'n_clicks'), + stream=True, + prevent_initial_call=True, +) +async def run(n): + yield 'Starting...' # replaces children immediately + async for token in llm(): + p = Patch() + p += token + yield p # appends to children (incremental) + yield 'Done' # last yield = final value +``` + +### Semantics + +- Each yield has the same shape as a regular return value (one value per + `Output`) and **replaces** the outputs. Yield `dash.Patch` objects for + incremental updates. +- `no_update` works per-output within a yield; a yield where nothing updates + produces no frame. Raising `PreventUpdate` mid-stream ends the stream + cleanly. +- `set_props()` between yields is folded into the next frame's `sideUpdate` + (HTTP) or streams immediately (WebSocket transport). +- Intermediate frames are applied through the same renderer path as + `set_props`, so dependent callbacks fire per frame and loading states stay + on until the stream completes (`Updating...` title for the whole stream). +- `on_error` applies per-stream: its return value becomes a final frame. + Without it, an exception mid-stream sends an error frame shown in devtools; + frames already applied stay applied. +- Works with sync generators (Flask/Quart/FastAPI) and async generators. + Sync generators warn at registration: they occupy a server worker (or WS + executor thread) for the whole stream — prefer `async def`. +- Incompatible with `background=True`, clientside callbacks, `mcp_enabled`, + and `api_endpoint` (validated at registration). + +### Transport & frame protocol + +Transport follows the callback's normal transport selection: if the callback +runs over the WebSocket callback transport (`websocket=True` or +`websocket_callbacks=True`), frames ride the open connection as +`callback_response` messages with `stream: true`; the terminal message is +`{status: 'ok', stream: true, done: true}`. Otherwise the HTTP POST response +streams NDJSON (`application/x-ndjson`), one frame per line: + +``` +{"multi": true, "response": {"": {"": }}, "sideUpdate": {...}?} +{"done": true} <- terminal frame +{"done": true, "error": {"message": "..."}} <- error terminal frame +``` + +The renderer applies each frame on arrival (via the `sideUpdate` path, so +`Patch` applies exactly once) and resolves the callback's execution promise +with an empty result on the terminal frame. + +### Caveats + +- Long streams should check `ctx.websocket.is_shutdown` (WS transport) in + loops; on HTTP, client disconnect raises `GeneratorExit` into the user + generator at its current `yield`. +- Proxies and compression middleware (nginx buffering, flask-compress/gzip, + Jupyter proxies) can buffer NDJSON and defeat streaming. Dash sets + `X-Accel-Buffering: no`, but middleware configuration may still be needed. +- Streamed frames bypass persistence (`prunePersistence`/`applyPersistence`). +- Wrapping a streamed output in `dcc.Loading` hides it for the entire stream + (loading stays on by design). +- Flask + async generator requires `dash[async]`; frames are bridged from a + private event-loop thread. +- `flask.request` inside a streamed callback body only works on the pure-WSGI + Flask path (no `dash[async]`/`use_async`); under async dispatch the request + context cannot be carried into the stream. Use Dash's `ctx` (cookies, + headers, args are captured at dispatch) instead. + +### Key Files + +- `dash/_callback.py` - `add_context_stream`/`async_add_context_stream` wrappers, frame builders +- `dash/_streaming.py` - `StreamedCallbackResponse` marker, context-safe iteration, NDJSON helpers +- `dash/backends/_flask.py`, `_quart.py`, `_fastapi.py` - streaming dispatch branches +- `dash/backends/ws.py` - `make_stream_frame_emitter`, `consume_stream_frames`/`aconsume_stream_frames` +- `dash/dash-renderer/src/actions/callbacks.ts` - `applyStreamFrame`, NDJSON reader, WS frame handling +- `dash/dash-renderer/src/utils/workerClient.ts` - stream-aware `callback_response` handling + ## Security ### XSS Protection diff --git a/CHANGELOG.md b/CHANGELOG.md index f3e7b816d7..d8444fa650 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ All notable changes to `dash` will be documented in this file. This project adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] + +### Added +- [#]() Streaming callbacks: `@callback(..., stream=True)` marks a generator (or async generator) callback whose yields are pushed to the browser as they are produced. Each yielded value has the same shape as a regular return value and replaces the outputs; yielding `dash.Patch` objects gives incremental updates (e.g. LLM token streaming). Streams ride the WebSocket callback transport when active, otherwise the HTTP response streams NDJSON frames. Works on Flask, Quart, and FastAPI; synchronous generators warn at registration since they occupy a server worker for the whole stream. + ## [4.4.0] - 2026-07-03 ### Added diff --git a/dash/_callback.py b/dash/_callback.py index 81a5345830..ac6aedf277 100644 --- a/dash/_callback.py +++ b/dash/_callback.py @@ -1,7 +1,9 @@ import collections import hashlib import inspect +import logging import warnings +from contextvars import copy_context from functools import wraps from typing import Callable, Optional, Any, List, Tuple, Union, Dict, TypeVar, cast @@ -22,6 +24,7 @@ MissingLongCallbackManagerError, BackgroundCallbackError, ImportedInsideCallbackError, + StreamCallbackError, ) from ._get_app import get_app from ._grouping import ( @@ -40,6 +43,7 @@ from .background_callback.managers import BaseBackgroundCallbackManager from ._callback_context import context_value +from ._streaming import StreamedCallbackResponse from .types import CallbackExecutionResponse from ._no_update import NoUpdate from . import _validate @@ -59,6 +63,8 @@ def _invoke_callback(func, *args, **kwargs): # used to mark the frame for the d return func(*args, **kwargs) # %% callback invoked %% +logger = logging.getLogger(__name__) + GLOBAL_CALLBACK_LIST: List[Any] = [] GLOBAL_CALLBACK_MAP: Dict[str, Any] = {} GLOBAL_INLINE_SCRIPTS: List[Any] = [] @@ -87,6 +93,7 @@ def callback( hidden: Optional[bool] = None, websocket: Optional[bool] = False, persistent: Optional[bool] = False, + stream: Optional[bool] = False, mcp_enabled: Optional[bool] = None, mcp_expose_docstring: Optional[bool] = None, **_kwargs, @@ -187,10 +194,34 @@ def callback( If True, this callback will not show the "Updating..." title while running. Useful for persistent WebSocket callbacks that stay active for long periods without requiring a loading indicator. + :param stream: + If True, the callback must be a generator (or async generator) + function. Each yielded value has the same shape as a regular + return value (one value per Output) and is pushed to the browser + immediately; yield `dash.Patch` objects for incremental updates. + Streams over the WebSocket callback transport when active, + otherwise over the HTTP response (NDJSON). Cannot be combined + with `background=True`. """ background_spec: Any = None + if stream: + if background: + raise BackgroundCallbackError( + "stream=True cannot be combined with background=True." + ) + if mcp_enabled: + raise StreamCallbackError( + "stream=True cannot be combined with mcp_enabled=True: " + "MCP tools expect a single JSON result." + ) + if api_endpoint: + raise StreamCallbackError( + "stream=True cannot be combined with api_endpoint: " + "API endpoints expect a single JSON result." + ) + config_prevent_initial_callbacks = _kwargs.pop( "config_prevent_initial_callbacks", False ) @@ -246,6 +277,7 @@ def callback( hidden=hidden, websocket=websocket, persistent=persistent, + stream=stream, mcp_enabled=mcp_enabled, mcp_expose_docstring=mcp_expose_docstring, ) @@ -301,6 +333,7 @@ def insert_callback( hidden=None, websocket=False, persistent=False, + stream=False, mcp_enabled=None, mcp_expose_docstring=None, ) -> str: @@ -330,6 +363,7 @@ def insert_callback( "hidden": hidden, "websocket": websocket, "persistent": persistent, + "stream": stream, } if running: callback_spec["running"] = running @@ -346,6 +380,7 @@ def insert_callback( "allow_dynamic_callbacks": dynamic_creator, "no_output": no_output, "websocket": websocket, + "stream": stream, "mcp_enabled": mcp_enabled, "mcp_expose_docstring": mcp_expose_docstring, } @@ -714,6 +749,7 @@ def register_callback( hidden=_kwargs.get("hidden", None), websocket=_kwargs.get("websocket", False), persistent=_kwargs.get("persistent", False), + stream=_kwargs.get("stream", False), mcp_enabled=_kwargs.get("mcp_enabled", None), mcp_expose_docstring=_kwargs.get("mcp_expose_docstring"), ) @@ -878,7 +914,219 @@ async def async_add_context(*args, **kwargs): return jsonResponse - if inspect.iscoroutinefunction(func): + def _build_stream_frame( + output_value, output_spec, callback_ctx, app, original_packages + ): + """Build one stream frame from a yielded value. + + Returns None when the yield produced no update (PreventUpdate or + all no_update with no set_props). + """ + response: CallbackExecutionResponse = {"multi": True} + try: + _prepare_response( + output_value, + output_spec, + multi, + response, + callback_ctx, + app, + original_packages, + None, + False, + has_output, + output, + callback_id, + allow_dynamic_callbacks, + ) + except PreventUpdate: + return None + finally: + # set_props between yields were folded into this frame's + # sideUpdate; don't resend them with later frames. + callback_ctx.updated_props.clear() + if not response.get("response") and not response.get("sideUpdate"): + return None + return response + + def _stream_error_frame(err): + logger.exception("Exception raised in streamed callback") + return {"done": True, "error": {"message": str(err) or repr(err)}} + + def _stream_frames( + user_gen, error_handler, output_spec, callback_ctx, app, original_packages + ): + try: + while True: + frame = None + try: + output_value = next(user_gen) + frame = _build_stream_frame( + output_value, + output_spec, + callback_ctx, + app, + original_packages, + ) + except (StopIteration, PreventUpdate): + break + except Exception as err: # pylint: disable=broad-exception-caught + if error_handler: + output_value = error_handler(err) + if output_value is not None: + frame = _build_stream_frame( + output_value, + output_spec, + callback_ctx, + app, + original_packages, + ) + if frame is not None: + yield frame + break + yield _stream_error_frame(err) + return + if frame is not None: + yield frame + yield {"done": True} + finally: + user_gen.close() + + async def _astream_frames( + user_gen, error_handler, output_spec, callback_ctx, app, original_packages + ): + # The whole stream is iterated from a single task (streaming + # response body or WS loop task), so setting the context var here + # makes ctx/set_props work for every step of the user generator. + token = context_value.set(callback_ctx) + try: + while True: + frame = None + try: + output_value = await user_gen.__anext__() + frame = _build_stream_frame( + output_value, + output_spec, + callback_ctx, + app, + original_packages, + ) + except (StopAsyncIteration, PreventUpdate): + break + except Exception as err: # pylint: disable=broad-exception-caught + if error_handler: + output_value = error_handler(err) + if output_value is not None: + frame = _build_stream_frame( + output_value, + output_spec, + callback_ctx, + app, + original_packages, + ) + if frame is not None: + yield frame + break + yield _stream_error_frame(err) + return + if frame is not None: + yield frame + yield {"done": True} + finally: + context_value.reset(token) + await user_gen.aclose() + + @wraps(func) + def add_context_stream(*args, **kwargs): + """Handles stream=True callbacks defined as sync generators.""" + error_handler = on_error or kwargs.pop("app_on_error", None) + + ( + output_spec, + callback_ctx, + func_args, + func_kwargs, + app, + original_packages, + _, + ) = _initialize_context( + args, kwargs, inputs_state_indices, has_output, insert_output + ) + + # Creates the generator; the function body runs on first next(). + user_gen = _invoke_callback(func, *func_args, **func_kwargs) + frames = _stream_frames( + user_gen, + error_handler, + output_spec, + callback_ctx, + app, + original_packages, + ) + # Snapshot the context (includes the callback context set above) so + # the transport can drive the generator after dispatch returns. + return StreamedCallbackResponse(frames, is_async=False, ctx=copy_context()) + + @wraps(func) + async def async_add_context_stream(*args, **kwargs): + """Handles stream=True callbacks defined as async generators.""" + error_handler = on_error or kwargs.pop("app_on_error", None) + + ( + output_spec, + callback_ctx, + func_args, + func_kwargs, + app, + original_packages, + _, + ) = _initialize_context( + args, kwargs, inputs_state_indices, has_output, insert_output + ) + + user_gen = _invoke_callback(func, *func_args, **func_kwargs) + frames = _astream_frames( + user_gen, + error_handler, + output_spec, + callback_ctx, + app, + original_packages, + ) + return StreamedCallbackResponse(frames, is_async=True) + + stream = _kwargs.get("stream", False) + is_gen_func = inspect.isgeneratorfunction(func) + is_async_gen_func = inspect.isasyncgenfunction(func) + if stream: + if not (is_gen_func or is_async_gen_func): + raise StreamCallbackError( + f"stream=True callback '{callback_id}' must be a generator " + "function (or async generator function) that yields output " + "updates." + ) + if is_gen_func: + # A sync generator stream occupies a server worker (or WS + # executor thread) for the whole stream duration; recommend + # async so it runs on the event loop. + warnings.warn( + f"stream=True callback '{callback_id}' is a synchronous " + "generator; it will occupy a server worker for the whole " + "stream. Define it with 'async def' so it runs on the " + "event loop instead.", + RuntimeWarning, + stacklevel=2, + ) + callback_map[callback_id]["callback"] = add_context_stream + else: + callback_map[callback_id]["callback"] = async_add_context_stream + elif is_gen_func or is_async_gen_func: + raise StreamCallbackError( + f"Callback '{callback_id}' is a generator function but was not " + "registered with stream=True. Did you mean " + "@callback(..., stream=True)?" + ) + elif inspect.iscoroutinefunction(func): callback_map[callback_id]["callback"] = async_add_context else: # A persistent, no-output callback streams via set_props and typically @@ -919,6 +1167,10 @@ def register_clientside_callback( *args, **kwargs, ): + if kwargs.get("stream"): + raise StreamCallbackError( + "stream=True is not supported for clientside callbacks." + ) output, inputs, state, prevent_initial_call = handle_callback_args(args, kwargs) no_output = isinstance(output, (list,)) and len(output) == 0 insert_callback( diff --git a/dash/_streaming.py b/dash/_streaming.py new file mode 100644 index 0000000000..d3a5609dbe --- /dev/null +++ b/dash/_streaming.py @@ -0,0 +1,189 @@ +"""Transport helpers for streaming callbacks (``@callback(..., stream=True)``). + +A streaming callback is a generator (or async generator) whose yields are +converted to "frames" — dicts with the same shape as a regular callback +response (see ``CallbackExecutionResponse``) — followed by a terminal +``{"done": True}`` frame. Frames are delivered to the renderer either as +NDJSON lines on the HTTP response or as individual messages over the +WebSocket callback transport. + +The frame generators are built in ``dash._callback``; this module owns the +marker object the backends dispatch on and the transport-side iteration +helpers. Iteration helpers exist because Dash's callback context lives in a +``contextvars.ContextVar`` and a sync generator runs in whatever context its +consumer drives it from — which, for a streaming HTTP response, is not the +request context the callback started in. +""" + +import asyncio +import contextlib +import functools +import logging +import queue +import threading +from typing import cast + +from ._utils import to_json as _to_json + +logger = logging.getLogger(__name__) + +STREAM_MIMETYPE = "application/x-ndjson" +# Disable proxy/server buffering so frames reach the browser as they are +# produced (X-Accel-Buffering covers nginx). +STREAM_HEADERS = {"X-Accel-Buffering": "no", "Cache-Control": "no-cache"} + +_SENTINEL = object() + + +def to_json(value) -> str: + return cast(str, _to_json(value)) + + +class StreamedCallbackResponse: # pylint: disable=too-few-public-methods + """Marker returned by ``stream=True`` callback wrappers. + + Backends detect this instead of a JSON string and return a streaming + response. ``frames`` is a generator (async generator when ``is_async``) + of frame dicts. ``ctx`` is the ``contextvars`` snapshot captured when the + callback was invoked; sync frame generators must be driven through it + (``iter_stream_frames``) so ``dash.ctx``/``set_props`` keep working after + the dispatch function has returned. + """ + + def __init__(self, frames, is_async, ctx=None): + self.frames = frames + self.is_async = is_async + self.ctx = ctx + + +def iter_stream_frames(marker): + """Drive a sync frame generator inside its captured context snapshot.""" + while True: + try: + yield marker.ctx.run(next, marker.frames) + except StopIteration: + return + + +async def aiter_stream_frames(marker): + """Async wrapper for a sync frame generator (ASGI backends). + + Each step runs on an executor thread through ``marker.ctx`` — Starlette's + own threadpool iteration would use a fresh context copy per chunk and + lose the callback context. + """ + loop = asyncio.get_running_loop() + while True: + frame = await loop.run_in_executor( + None, marker.ctx.run, functools.partial(next, marker.frames, _SENTINEL) + ) + if frame is _SENTINEL: + return + yield frame + + +def _serialize_frame(frame): + """Serialize one frame to an NDJSON line. + + Returns ``(line, fatal)``; a serialization failure produces a terminal + error frame so the client is not left waiting on a silently dead stream. + """ + try: + return to_json(frame) + "\n", False + except TypeError as err: + logger.exception("Failed to serialize streamed callback frame") + return ( + to_json( + { + "done": True, + "error": { + "message": "Non-serializable value in streamed " + f"callback output: {err}" + }, + } + ) + + "\n", + True, + ) + + +def ndjson_lines(marker): + """Sync NDJSON body for a sync frame generator (Flask/WSGI).""" + for frame in iter_stream_frames(marker): + line, fatal = _serialize_frame(frame) + yield line + if fatal: + return + + +async def andjson_lines(frames): + """Async NDJSON body over an async iterator of frames.""" + async for frame in frames: + line, fatal = _serialize_frame(frame) + yield line + if fatal: + return + + +def marker_ndjson_aiter(marker): + """Async NDJSON body for either flavor of frame generator.""" + if marker.is_async: + return andjson_lines(marker.frames) + return andjson_lines(aiter_stream_frames(marker)) + + +def sync_iter_asyncgen(agen): + """Iterate an async generator from sync code (Flask + async gen). + + Runs the whole consumption on one task on a private event-loop thread so + contextvars set inside the generator persist across steps. Closing this + generator (client disconnect) cancels the task, which raises into the + user generator at its current yield. + """ + frame_queue: queue.Queue = queue.Queue() + loop = asyncio.new_event_loop() + task = None + task_ready = threading.Event() + + async def consume(): + try: + async for item in agen: + frame_queue.put(("item", item)) + frame_queue.put(("end", None)) + except BaseException as err: # pylint: disable=broad-exception-caught + frame_queue.put(("error", err)) + finally: + with contextlib.suppress(Exception): + await agen.aclose() + + def run(): + nonlocal task + asyncio.set_event_loop(loop) + task = loop.create_task(consume()) + task_ready.set() + try: + loop.run_until_complete(task) + except BaseException: # pylint: disable=broad-exception-caught + pass + finally: + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.close() + + thread = threading.Thread(target=run, daemon=True, name="dash-stream-bridge") + thread.start() + try: + while True: + kind, value = frame_queue.get() + if kind == "item": + yield value + elif kind == "error": + if isinstance(value, asyncio.CancelledError): + return + raise value + else: + return + finally: + task_ready.wait(timeout=5) + if task is not None and not task.done(): + with contextlib.suppress(RuntimeError): + loop.call_soon_threadsafe(task.cancel) diff --git a/dash/backends/_fastapi.py b/dash/backends/_fastapi.py index 6a23c4873b..669a6e7bda 100644 --- a/dash/backends/_fastapi.py +++ b/dash/backends/_fastapi.py @@ -20,7 +20,7 @@ try: from fastapi import FastAPI, Request, Response, Body - from fastapi.responses import JSONResponse, RedirectResponse + from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse from fastapi.staticfiles import StaticFiles from starlette.responses import Response as StarletteResponse from starlette.datastructures import MutableHeaders @@ -36,6 +36,12 @@ from dash.fingerprint import check_fingerprint from dash import _validate, get_app +from dash._streaming import ( + STREAM_HEADERS, + STREAM_MIMETYPE, + StreamedCallbackResponse, + marker_ndjson_aiter, +) from dash.exceptions import PreventUpdate from .base_server import ( BaseDashServer, @@ -48,6 +54,7 @@ run_callback_in_executor, run_callback_on_loop, make_callback_done_handler, + make_stream_frame_emitter, shutdown_ws_connection, ) from ._utils import format_traceback_html @@ -566,6 +573,12 @@ async def _dispatch(request: Request): # pylint: disable=unused-argument response_data = ctx.run(partial_func) if inspect.iscoroutine(response_data): response_data = await response_data + if isinstance(response_data, StreamedCallbackResponse): + return StreamingResponse( + marker_ndjson_aiter(response_data), + media_type=STREAM_MIMETYPE, + headers=dict(STREAM_HEADERS), + ) return cb_ctx.dash_response.set_response(data=response_data) return _dispatch @@ -812,6 +825,12 @@ async def websocket_handler(websocket: WebSocket): renderer_id, shutdown_event, ) + stream_emitter = make_stream_frame_emitter( + outbound_queue, + request_id, + renderer_id, + shutdown_event, + ) if is_async: task = asyncio.create_task( @@ -820,6 +839,7 @@ async def websocket_handler(websocket: WebSocket): payload, ws_cb, FastAPIResponseAdapter(), + stream_emitter, ) ) task.add_done_callback(done_handler) @@ -832,6 +852,7 @@ async def websocket_handler(websocket: WebSocket): payload, ws_cb, FastAPIResponseAdapter(), + stream_emitter, ) # Set up done callback to send response future.add_done_callback(done_handler) diff --git a/dash/backends/_flask.py b/dash/backends/_flask.py index 00c8730d8a..171cc83f36 100644 --- a/dash/backends/_flask.py +++ b/dash/backends/_flask.py @@ -22,6 +22,7 @@ g as flask_g, has_request_context, redirect, + stream_with_context, ) from werkzeug.debug import tbtools @@ -29,6 +30,14 @@ from dash import _validate from dash.exceptions import PreventUpdate, InvalidResourceError from dash._callback import _invoke_callback, _async_invoke_callback +from dash._streaming import ( + STREAM_HEADERS, + STREAM_MIMETYPE, + StreamedCallbackResponse, + andjson_lines, + ndjson_lines, + sync_iter_asyncgen, +) from dash._utils import parse_version from .base_server import BaseDashServer, RequestAdapter, ResponseAdapter @@ -251,6 +260,30 @@ def add_redirect_rule(self, app, fullname, path): # pylint: disable=unused-argument def serve_callback(self, dash_app: Dash): + def _stream_response( + marker: StreamedCallbackResponse, with_request_ctx: bool + ) -> Response: + if marker.is_async: + # Drive the async frame generator on a private event-loop + # thread; the response iterator drains it synchronously. + body = sync_iter_asyncgen(andjson_lines(marker.frames)) + else: + body = ndjson_lines(marker) + if with_request_ctx: + # Keep flask.request usable while the body is iterated (the + # generator runs after the view returns). Only valid on the + # sync dispatch path: under an async view (asgiref) the + # request context lives in a different contextvars context + # and stream_with_context would corrupt its teardown. Dash's + # own callback context always works — it travels in the + # marker's context snapshot. + body = stream_with_context(body) + return Response( + body, + content_type=STREAM_MIMETYPE, + headers=dict(STREAM_HEADERS), + ) + def _dispatch(): body = request.get_json() # pylint: disable=protected-access @@ -262,6 +295,8 @@ def _dispatch(): func, args, cb_ctx.outputs_list, cb_ctx ) response_data = ctx.run(partial_func) + if isinstance(response_data, StreamedCallbackResponse): + return _stream_response(response_data, with_request_ctx=True) if asyncio.iscoroutine(response_data): raise Exception( "You are trying to use a coroutine without dash[async]. " @@ -283,6 +318,8 @@ async def _dispatch_async(): response_data = ctx.run(partial_func) if asyncio.iscoroutine(response_data): response_data = await response_data + if isinstance(response_data, StreamedCallbackResponse): + return _stream_response(response_data, with_request_ctx=False) return cb_ctx.dash_response.set_response(data=response_data) if dash_app._use_async: # pylint: disable=protected-access diff --git a/dash/backends/_quart.py b/dash/backends/_quart.py index dde98061d1..97c9fca31b 100644 --- a/dash/backends/_quart.py +++ b/dash/backends/_quart.py @@ -40,6 +40,12 @@ from dash.exceptions import PreventUpdate, InvalidResourceError from dash.fingerprint import check_fingerprint +from dash._streaming import ( + STREAM_HEADERS, + STREAM_MIMETYPE, + StreamedCallbackResponse, + marker_ndjson_aiter, +) from dash._utils import parse_version from dash import _validate from .base_server import ( @@ -53,6 +59,7 @@ run_callback_in_executor, run_callback_on_loop, make_callback_done_handler, + make_stream_frame_emitter, shutdown_ws_connection, ) from ._utils import format_traceback_html @@ -400,6 +407,12 @@ async def _dispatch(): response_data = ctx.run(partial_func) if inspect.iscoroutine(response_data): # if user callback is async response_data = await response_data + if isinstance(response_data, StreamedCallbackResponse): + return Response( # type: ignore[return-value] + marker_ndjson_aiter(response_data), + content_type=STREAM_MIMETYPE, + headers=dict(STREAM_HEADERS), + ) return cb_ctx.dash_response.set_response(data=response_data) # type: ignore[arg-type] return _dispatch @@ -643,6 +656,12 @@ async def websocket_handler(): # pylint: disable=too-many-branches renderer_id, connection_shutdown_event, ) + stream_emitter = make_stream_frame_emitter( + outbound_queue, + request_id, + renderer_id, + connection_shutdown_event, + ) if is_async: task = asyncio.create_task( @@ -651,6 +670,7 @@ async def websocket_handler(): # pylint: disable=too-many-branches payload, ws_cb, QuartResponseAdapter(), + stream_emitter, ) ) task.add_done_callback(done_handler) @@ -663,6 +683,7 @@ async def websocket_handler(): # pylint: disable=too-many-branches payload, ws_cb, QuartResponseAdapter(), + stream_emitter, ) # Set up done callback to send response future.add_done_callback(done_handler) diff --git a/dash/backends/ws.py b/dash/backends/ws.py index 6dc2c0e926..2f74851e34 100644 --- a/dash/backends/ws.py +++ b/dash/backends/ws.py @@ -21,6 +21,12 @@ from dash.exceptions import PreventUpdate, WebsocketDisconnected from dash.types import CallbackExecutionBody +from dash._streaming import ( + StreamedCallbackResponse, + aiter_stream_frames, + iter_stream_frames, + sync_iter_asyncgen, +) from dash._utils import to_json if TYPE_CHECKING: @@ -506,6 +512,105 @@ def on_done(f: "concurrent.futures.Future | asyncio.Future") -> None: return on_done +def make_stream_frame_emitter( + outbound_queue: janus.Queue[str], + request_id: str, + renderer_id: str, + shutdown_event: threading.Event, +) -> Callable[[dict], None]: + """Create an emitter sending intermediate stream frames for a request. + + Frames ride the ``callback_response`` message type with ``stream: True`` + and no ``done`` flag, so the renderer keeps the request pending until the + final ``callback_response`` (sent by the done handler) resolves it. + """ + + def emit(frame: dict) -> None: + if shutdown_event.is_set(): + return + outbound_queue.sync_q.put_nowait( + cast( + str, + to_json( + { + "type": "callback_response", + "rendererId": renderer_id, + "requestId": request_id, + "payload": {"status": "ok", "stream": True, "data": frame}, + } + ), + ) + ) + # Flush so the frame doesn't sit in the sender's batching window. + outbound_queue.sync_q.put_nowait(FLUSH_SIGNAL) + + return emit + + +_STREAM_DONE_PAYLOAD = {"status": "ok", "stream": True, "done": True} + + +def _stream_frame_result(frame: dict) -> "dict | None": + """Terminal payload for a frame, or None if it is an intermediate frame.""" + if not frame.get("done"): + return None + error = frame.get("error") + if error: + return {"status": "error", "message": error.get("message", "")} + return dict(_STREAM_DONE_PAYLOAD) + + +def consume_stream_frames( + marker: StreamedCallbackResponse, + ws_callback: DashWebsocketCallback, + stream_emitter: "Callable[[dict], None] | None", +) -> dict: + """Drain a streamed callback synchronously (threadpool path). + + Emits intermediate frames over the WebSocket and returns the terminal + payload for the done handler to send as the final callback_response. + """ + emit = stream_emitter or (lambda _frame: None) + if marker.is_async: + # Defensive: an async generator that ended up on the thread path is + # driven on a private event-loop thread. + frames = sync_iter_asyncgen(marker.frames) + else: + frames = iter_stream_frames(marker) + try: + for frame in frames: + if ws_callback.is_shutdown: + return {"status": "prevent_update"} + result = _stream_frame_result(frame) + if result is not None: + return result + emit(frame) + return dict(_STREAM_DONE_PAYLOAD) + finally: + frames.close() + + +async def aconsume_stream_frames( + marker: StreamedCallbackResponse, + ws_callback: DashWebsocketCallback, + stream_emitter: "Callable[[dict], None] | None", +) -> dict: + """Drain a streamed callback on the event loop (async dispatch path).""" + emit = stream_emitter or (lambda _frame: None) + frames = marker.frames if marker.is_async else aiter_stream_frames(marker) + try: + async for frame in frames: + if ws_callback.is_shutdown: + return {"status": "prevent_update"} + result = _stream_frame_result(frame) + if result is not None: + return result + emit(frame) + return dict(_STREAM_DONE_PAYLOAD) + finally: + await frames.aclose() + + def _prepare_ws_partial( dash_app: "dash.Dash", payload: CallbackExecutionBody, @@ -533,6 +638,7 @@ def run_callback_in_executor( payload: CallbackExecutionBody, ws_callback: DashWebsocketCallback, response_adapter: "ResponseAdapter", + stream_emitter: "Callable[[dict], None] | None" = None, ) -> concurrent.futures.Future: """Submit a synchronous callback to the executor for thread pool execution. @@ -571,6 +677,10 @@ def run_callback(): return result response_data = ctx.run(run_callback) + if isinstance(response_data, StreamedCallbackResponse): + # The frame generator carries its own context snapshot, so it + # can be driven outside ctx here. + return consume_stream_frames(response_data, ws_callback, stream_emitter) return {"status": "ok", "data": json.loads(response_data)} except PreventUpdate: @@ -589,6 +699,7 @@ async def run_callback_on_loop( payload: CallbackExecutionBody, ws_callback: DashWebsocketCallback, response_adapter: "ResponseAdapter", + stream_emitter: "Callable[[dict], None] | None" = None, ) -> dict: """Run an async callback as a task on the connection's event loop. @@ -616,6 +727,10 @@ async def run_callback_on_loop( ) result = partial_func() response_data = await result if inspect.iscoroutine(result) else result + if isinstance(response_data, StreamedCallbackResponse): + return await aconsume_stream_frames( + response_data, ws_callback, stream_emitter + ) return {"status": "ok", "data": json.loads(response_data)} except PreventUpdate: diff --git a/dash/dash-renderer/src/actions/callbacks.ts b/dash/dash-renderer/src/actions/callbacks.ts index e6604a2337..027601fc09 100644 --- a/dash/dash-renderer/src/actions/callbacks.ts +++ b/dash/dash-renderer/src/actions/callbacks.ts @@ -460,6 +460,36 @@ function sideUpdate(outputs: SideUpdateOutput, cb: ICallbackPayload) { }; } +/** + * Apply an intermediate frame from a stream=True callback. + * + * The frame's declared outputs are flattened to `id.prop` keys and applied + * through the sideUpdate path (parsePatchProps + updateComponent), so Patch + * values apply incrementally, paths recompute for newly added components, + * and observers are notified — the same semantics as set_props. The + * callback's execution promise stays pending until the terminal frame. + */ +function applyStreamFrame( + dispatch: any, + frame: CallbackResponseData, + payload: ICallbackPayload +) { + if (frame.response) { + const flat: SideUpdateOutput = {}; + toPairs(frame.response).forEach(([id, props]) => { + toPairs(props as Record).forEach(([prop, value]) => { + flat[`${id}.${prop}`] = value; + }); + }); + if (keys(flat).length) { + dispatch(sideUpdate(flat, payload)); + } + } + if (frame.sideUpdate) { + dispatch(sideUpdate(frame.sideUpdate, payload)); + } +} + function handleServerside( dispatch: any, hooks: any, @@ -609,7 +639,92 @@ function handleServerside( } }; + // stream=True callbacks: read NDJSON frames as they arrive, + // apply each one immediately, and resolve on the terminal frame. + const handleStreamedResponse = async (streamRes: any) => { + const reader = streamRes.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + let lastResponse: CallbackResponse | undefined; + let finished = false; + + const processFrame = async (frame: CallbackResponseData) => { + if (frame.done) { + finished = true; + completeJob(); + if (frame.error) { + recordProfile({}); + reject( + new Error( + frame.error.message || 'Callback error' + ) + ); + return; + } + if (hooks.request_post) { + hooks.request_post(payload, lastResponse); + } + recordProfile(lastResponse || {}); + // Frames were already applied; resolve empty so the + // terminal store update is a no-op (Patch frames must + // not re-apply). + resolve({}); + return; + } + if (frame.dist) { + await Promise.all(frame.dist.map(loadLibrary)); + } + lastResponse = frame.response; + applyStreamFrame(dispatch, frame, payload); + }; + + try { + for (;;) { + const {done, value} = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, {stream: true}); + const lines = buffer.split('\n'); + buffer = lines.pop() as string; + for (const line of lines) { + if (!line.trim()) { + continue; + } + await processFrame(JSON.parse(line)); + if (finished) { + reader.cancel(); + return; + } + } + } + if (buffer.trim() && !finished) { + await processFrame(JSON.parse(buffer)); + } + } catch (err) { + if (!finished) { + finished = true; + completeJob(); + recordProfile({}); + reject(err); + return; + } + } + if (!finished) { + // Stream ended without a terminal frame (connection + // dropped): clear loading, keep applied frames. + completeJob(); + recordProfile(lastResponse || {}); + resolve({}); + } + }; + if (status === STATUS.OK) { + const contentType = res.headers.get('Content-Type') || ''; + if (contentType.includes('application/x-ndjson') && res.body) { + handleStreamedResponse(res); + return; + } res.json().then((data: CallbackResponseData) => { if (!cacheKey && data.cacheKey) { cacheKey = data.cacheKey; @@ -721,7 +836,16 @@ async function handleWebsocketCallback( // Ensure WebSocket connection is established await workerClient.ensureConnected(config); - const response = await workerClient.sendCallback(payload); + // stream=True callbacks deliver intermediate frames before the + // terminal response; apply each one as it arrives. + let lastStreamResponse: CallbackResponse | undefined; + const response = await workerClient.sendCallback( + payload, + (frame: CallbackResponseData) => { + lastStreamResponse = frame?.response; + applyStreamFrame(dispatch, frame, payload); + } + ); // Handle running off state if (runningOff) { @@ -755,6 +879,34 @@ async function handleWebsocketCallback( throw new Error(response.message || 'Callback error'); } + if (response.stream) { + // Terminal frame of a streamed callback: every frame was already + // applied on arrival, so resolve empty (the terminal store update + // must be a no-op — Patch frames must not re-apply). + if (hooks.request_post) { + hooks.request_post(payload, lastStreamResponse); + } + if (config.ui) { + const totalTime = Date.now() - requestTime; + dispatch( + updateResourceUsage({ + id: payload.output, + usage: { + __dash_server: totalTime, + __dash_client: totalTime, + __dash_upload: 0, + __dash_download: 0 + }, + status: STATUS.OK, + result: lastStreamResponse || {}, + inputs: payload.inputs, + state: payload.state + }) + ); + } + return {}; + } + // Extract the callback data - structure is {multi: boolean, response: {...}} const callbackData = response.data as CallbackResponseData; diff --git a/dash/dash-renderer/src/types/callbacks.ts b/dash/dash-renderer/src/types/callbacks.ts index 5f963463d2..d674c69f3a 100644 --- a/dash/dash-renderer/src/types/callbacks.ts +++ b/dash/dash-renderer/src/types/callbacks.ts @@ -17,6 +17,7 @@ export interface ICallbackDefinition { no_output?: boolean; websocket?: boolean; persistent?: boolean; + stream?: boolean; } export interface ICallbackProperty { @@ -109,6 +110,9 @@ export type CallbackResponseData = { cancel?: ICallbackProperty[]; dist?: any; sideUpdate?: any; + // stream=True callbacks: terminal frame marker and mid-stream error. + done?: boolean; + error?: {message?: string}; }; export type SideUpdateOutput = { diff --git a/dash/dash-renderer/src/utils/workerClient.ts b/dash/dash-renderer/src/utils/workerClient.ts index 1c07a4c1c3..31943d7ef8 100644 --- a/dash/dash-renderer/src/utils/workerClient.ts +++ b/dash/dash-renderer/src/utils/workerClient.ts @@ -25,6 +25,10 @@ export interface CallbackResponse { status: 'ok' | 'prevent_update' | 'error'; data?: Record; message?: string; + /** True for responses from a stream=True callback */ + stream?: boolean; + /** True on the terminal frame of a streamed callback */ + done?: boolean; } /** Set props message payload */ @@ -43,6 +47,8 @@ export interface GetPropsRequestPayload { interface PendingRequest { resolve: (value: CallbackResponse) => void; reject: (error: Error) => void; + /** Receives intermediate frames from a stream=True callback */ + onFrame?: (data: Record) => void; } /** @@ -203,9 +209,15 @@ class WorkerClient { /** * Send a callback request to the server via the worker. * @param payload The callback payload + * @param onFrame Optional handler for intermediate frames from a + * stream=True callback; the returned promise still resolves once with + * the terminal response. * @returns Promise that resolves with the callback response */ - public async sendCallback(payload: unknown): Promise { + public async sendCallback( + payload: unknown, + onFrame?: (data: Record) => void + ): Promise { // Wait for initial connection if one is in progress if (this.connectionPromise && !this.isConnected) { await this.connectionPromise; @@ -218,7 +230,7 @@ class WorkerClient { const requestId = `${this.rendererId}-${++this.requestCounter}`; return new Promise((resolve, reject) => { - this.pendingCallbacks.set(requestId, {resolve, reject}); + this.pendingCallbacks.set(requestId, {resolve, reject, onFrame}); this.worker!.port.postMessage({ type: WorkerMessageType.CALLBACK_REQUEST, @@ -301,8 +313,21 @@ class WorkerClient { const requestId = message.requestId; const pending = this.pendingCallbacks.get(requestId); if (pending) { + const payload = message.payload; + if ( + payload?.stream && + !payload.done && + payload.status === 'ok' + ) { + // Intermediate stream frame: deliver it and keep the + // request pending until the terminal frame arrives. + if (pending.onFrame) { + pending.onFrame(payload.data); + } + break; + } this.pendingCallbacks.delete(requestId); - pending.resolve(message.payload); + pending.resolve(payload); } break; } diff --git a/dash/exceptions.py b/dash/exceptions.py index 9366f9359c..9d04caece8 100644 --- a/dash/exceptions.py +++ b/dash/exceptions.py @@ -121,3 +121,7 @@ class WebSocketCallbackError(CallbackException): class WebsocketDisconnected(CallbackException): pass + + +class StreamCallbackError(CallbackException): + pass diff --git a/dash/types.py b/dash/types.py index 9da246b16c..af9b66de3f 100644 --- a/dash/types.py +++ b/dash/types.py @@ -102,3 +102,6 @@ class CallbackExecutionResponse(TypedDict): response: NotRequired[Dict[str, CallbackOutput]] sideUpdate: NotRequired[Dict[str, CallbackSideOutput]] dist: NotRequired[List[Any]] + # stream=True callbacks: terminal frame marker and mid-stream error. + done: NotRequired[bool] + error: NotRequired[Dict[str, str]] diff --git a/tests/integration/callbacks/test_stream_callbacks.py b/tests/integration/callbacks/test_stream_callbacks.py new file mode 100644 index 0000000000..aa5edf75f4 --- /dev/null +++ b/tests/integration/callbacks/test_stream_callbacks.py @@ -0,0 +1,217 @@ +"""Browser integration tests for stream=True callbacks over HTTP (NDJSON).""" +import time + +import pytest + +from dash import ( + Dash, + Input, + Output, + Patch, + html, + no_update, + set_props, +) +from dash.testing.wait import until + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stst001_stream_progressive_render(dash_duo): + """Intermediate yields render before the stream completes.""" + app = Dash(__name__) + app.layout = html.Div( + [ + html.Button("Start", id="btn", n_clicks=0), + html.Div(id="out", children="idle"), + ] + ) + + @app.callback( + Output("out", "children"), + Input("btn", "n_clicks"), + stream=True, + prevent_initial_call=True, + ) + def stream_cb(n): + yield "step-1" + time.sleep(0.5) + yield "step-2" + time.sleep(0.5) + yield "done" + + dash_duo.start_server(app) + dash_duo.find_element("#btn").click() + # Each yield renders while the callback is still running. + dash_duo.wait_for_text_to_equal("#out", "step-1") + dash_duo.wait_for_text_to_equal("#out", "step-2") + dash_duo.wait_for_text_to_equal("#out", "done") + assert dash_duo.get_logs() == [] + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stst002_stream_patch_appends_once(dash_duo): + """Patch yields apply exactly once (token streaming).""" + app = Dash(__name__) + app.layout = html.Div( + [ + html.Button("Start", id="btn", n_clicks=0), + html.Div(id="out", children=""), + ] + ) + + @app.callback( + Output("out", "children"), + Input("btn", "n_clicks"), + stream=True, + prevent_initial_call=True, + ) + def stream_cb(n): + yield "->" + for token in ["alpha", "beta", "gamma"]: + time.sleep(0.2) + patch = Patch() + patch += token + yield patch + + dash_duo.start_server(app) + dash_duo.find_element("#btn").click() + # Exact concatenation catches both double-apply and dropped frames. + dash_duo.wait_for_text_to_equal("#out", "->alphabetagamma") + # Give any straggler updates a chance to (incorrectly) re-apply. + time.sleep(0.5) + assert dash_duo.find_element("#out").text == "->alphabetagamma" + assert dash_duo.get_logs() == [] + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stst003_stream_multi_output_and_set_props(dash_duo): + app = Dash(__name__) + app.layout = html.Div( + [ + html.Button("Start", id="btn", n_clicks=0), + html.Div(id="a", children=""), + html.Div(id="b", children=""), + html.Div(id="side", children=""), + ] + ) + + @app.callback( + Output("a", "children"), + Output("b", "children"), + Input("btn", "n_clicks"), + stream=True, + prevent_initial_call=True, + ) + def stream_cb(n): + yield "a1", no_update + time.sleep(0.3) + set_props("side", {"children": "from-set-props"}) + yield no_update, "b1" + + dash_duo.start_server(app) + dash_duo.find_element("#btn").click() + dash_duo.wait_for_text_to_equal("#a", "a1") + dash_duo.wait_for_text_to_equal("#b", "b1") + dash_duo.wait_for_text_to_equal("#side", "from-set-props") + assert dash_duo.get_logs() == [] + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stst004_stream_triggers_downstream_callback(dash_duo): + """The final streamed value triggers dependent callbacks.""" + app = Dash(__name__) + app.layout = html.Div( + [ + html.Button("Start", id="btn", n_clicks=0), + html.Div(id="out", children=""), + html.Div(id="downstream", children=""), + ] + ) + + @app.callback( + Output("out", "children"), + Input("btn", "n_clicks"), + stream=True, + prevent_initial_call=True, + ) + def stream_cb(n): + yield "one" + time.sleep(0.2) + yield "two" + + @app.callback( + Output("downstream", "children"), + Input("out", "children"), + prevent_initial_call=True, + ) + def downstream(value): + return f"saw: {value}" + + dash_duo.start_server(app) + dash_duo.find_element("#btn").click() + dash_duo.wait_for_text_to_equal("#downstream", "saw: two") + assert dash_duo.get_logs() == [] + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stst005_stream_error_shows_in_devtools(dash_duo): + app = Dash(__name__) + app.layout = html.Div( + [ + html.Button("Start", id="btn", n_clicks=0), + html.Div(id="out", children=""), + ] + ) + + @app.callback( + Output("out", "children"), + Input("btn", "n_clicks"), + stream=True, + prevent_initial_call=True, + ) + def stream_cb(n): + yield "before-error" + raise ValueError("stream blew up") + + dash_duo.start_server(app, debug=True, use_reloader=False, use_debugger=True) + dash_duo.find_element("#btn").click() + # The frame before the error stays applied. + dash_duo.wait_for_text_to_equal("#out", "before-error") + # And the error surfaces in the devtools error count. + dash_duo.wait_for_text_to_equal(".test-devtools-error-count", "1") + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stst006_stream_loading_state(dash_duo): + """The callback stays in loading state for the whole stream.""" + app = Dash(__name__) + app.layout = html.Div( + [ + html.Button("Start", id="btn", n_clicks=0), + html.Div(id="out", children="idle"), + ] + ) + + @app.callback( + Output("out", "children"), + Input("btn", "n_clicks"), + stream=True, + prevent_initial_call=True, + ) + def stream_cb(n): + yield "working" + time.sleep(1.5) + yield "finished" + + dash_duo.start_server(app) + dash_duo.find_element("#btn").click() + dash_duo.wait_for_text_to_equal("#out", "working") + # An intermediate frame rendered but the callback is still running: + # the loading state stays on (document title shows "Updating..."). + until(lambda: dash_duo.driver.title == "Updating...", timeout=3) + assert dash_duo.redux_state_is_loading + dash_duo.wait_for_text_to_equal("#out", "finished") + # After the terminal frame the loading state clears. + until(lambda: dash_duo.driver.title != "Updating...", timeout=3) + assert not dash_duo.redux_state_is_loading + assert dash_duo.get_logs() == [] diff --git a/tests/unit/test_stream_callbacks.py b/tests/unit/test_stream_callbacks.py new file mode 100644 index 0000000000..117e831373 --- /dev/null +++ b/tests/unit/test_stream_callbacks.py @@ -0,0 +1,283 @@ +"""Unit tests for stream=True callbacks - no browser required.""" +import asyncio +import contextvars +import json + +import pytest + +from dash import Dash, Input, Output, Patch, callback, html, no_update, set_props +from dash._callback import GLOBAL_CALLBACK_LIST, GLOBAL_CALLBACK_MAP +from dash._streaming import sync_iter_asyncgen +from dash.exceptions import ( + BackgroundCallbackError, + PreventUpdate, + StreamCallbackError, +) + + +def make_body(output_id, prop, input_id="btn"): + return { + "output": f"{output_id}.{prop}", + "outputs": {"id": output_id, "property": prop}, + "inputs": [{"id": input_id, "property": "n_clicks", "value": 1}], + "changedPropIds": [f"{input_id}.n_clicks"], + } + + +def post_stream(app, body): + """POST a callback request and return the parsed NDJSON frames.""" + client = app.server.test_client() + resp = client.post("/_dash-update-component", json=body) + assert resp.status_code == 200 + assert resp.headers.get("Content-Type") == "application/x-ndjson" + data = resp.get_data(as_text=True) + return [json.loads(line) for line in data.splitlines() if line.strip()] + + +def test_stcb001_stream_requires_generator(): + with pytest.raises(StreamCallbackError, match="must be a generator"): + + @callback(Output("stcb001", "children"), Input("in", "value"), stream=True) + def not_a_generator(value): + return value + + +def test_stcb002_generator_requires_stream(): + with pytest.raises(StreamCallbackError, match="stream=True"): + + @callback(Output("stcb002", "children"), Input("in", "value")) + def a_generator(value): + yield value + + +def test_stcb003_stream_incompatible_kwargs(): + with pytest.raises(BackgroundCallbackError): + + @callback( + Output("stcb003a", "children"), + Input("in", "value"), + stream=True, + background=True, + ) + def bg(value): + yield value + + with pytest.raises(StreamCallbackError, match="mcp_enabled"): + + @callback( + Output("stcb003b", "children"), + Input("in", "value"), + stream=True, + mcp_enabled=True, + ) + def mcp(value): + yield value + + with pytest.raises(StreamCallbackError, match="api_endpoint"): + + @callback( + Output("stcb003c", "children"), + Input("in", "value"), + stream=True, + api_endpoint="/stream", + ) + def api(value): + yield value + + +def test_stcb004_stream_not_clientside(): + app = Dash(__name__) + with pytest.raises(StreamCallbackError, match="clientside"): + app.clientside_callback( + "function(v) { return v; }", + Output("stcb004", "children"), + Input("in", "value"), + stream=True, + ) + + +def test_stcb005_sync_generator_warns(): + with pytest.warns(RuntimeWarning, match="synchronous generator"): + + @callback(Output("stcb005", "children"), Input("in", "value"), stream=True) + def sync_gen(value): + yield value + + +def test_stcb006_async_generator_no_warning(recwarn): + @callback(Output("stcb006", "children"), Input("in", "value"), stream=True) + async def async_gen(value): + yield value + + assert not [w for w in recwarn.list if issubclass(w.category, RuntimeWarning)] + + +def test_stcb007_stream_flag_in_spec_and_map(): + @callback(Output("stcb007", "children"), Input("in", "value"), stream=True) + async def async_gen(value): + yield value + + spec = [s for s in GLOBAL_CALLBACK_LIST if s["output"] == "stcb007.children"][-1] + assert spec["stream"] is True + assert GLOBAL_CALLBACK_MAP["stcb007.children"]["stream"] is True + + @callback(Output("stcb007b", "children"), Input("in", "value")) + def regular(value): + return value + + spec = [s for s in GLOBAL_CALLBACK_LIST if s["output"] == "stcb007b.children"][-1] + assert spec["stream"] is False + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stcb008_flask_ndjson_frames(): + app = Dash(__name__) + app.layout = html.Div( + [html.Button(id="btn"), html.Div(id="out"), html.Div(id="side")] + ) + + @app.callback(Output("out", "children"), Input("btn", "n_clicks"), stream=True) + def stream_cb(n): + yield "start" + patch = Patch() + patch += " token" + set_props("side", {"children": "side-value"}) + yield patch + yield "final" + + frames = post_stream(app, make_body("out", "children")) + assert frames[0] == {"multi": True, "response": {"out": {"children": "start"}}} + # Patch value serialized with set_props folded into the same frame, + # and cleared so it is not resent with the next frame. + assert frames[1]["sideUpdate"] == {"side": {"children": "side-value"}} + assert ( + frames[1]["response"]["out"]["children"]["__dash_patch_update"] + == "__dash_patch_update" + ) + assert frames[2] == {"multi": True, "response": {"out": {"children": "final"}}} + assert frames[3] == {"done": True} + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stcb009_stream_error_frame(): + app = Dash(__name__) + app.layout = html.Div([html.Button(id="btn"), html.Div(id="out")]) + + @app.callback(Output("out", "children"), Input("btn", "n_clicks"), stream=True) + def err_cb(n): + yield "one" + raise ValueError("boom") + + frames = post_stream(app, make_body("out", "children")) + assert frames[0]["response"] == {"out": {"children": "one"}} + assert frames[1]["done"] is True + assert "boom" in frames[1]["error"]["message"] + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stcb010_stream_on_error_handler(): + app = Dash(__name__) + app.layout = html.Div([html.Button(id="btn"), html.Div(id="out")]) + + def handle(err): + return f"handled: {err}" + + @app.callback( + Output("out", "children"), + Input("btn", "n_clicks"), + stream=True, + on_error=handle, + ) + def err_cb(n): + yield "one" + raise ValueError("boom") + + frames = post_stream(app, make_body("out", "children")) + assert frames[0]["response"] == {"out": {"children": "one"}} + assert frames[1]["response"] == {"out": {"children": "handled: boom"}} + assert frames[2] == {"done": True} + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stcb011_prevent_update_and_no_update_yields(): + app = Dash(__name__) + app.layout = html.Div( + [html.Button(id="btn"), html.Div(id="out"), html.Div(id="out2")] + ) + + @app.callback( + Output("out", "children"), + Output("out2", "children"), + Input("btn", "n_clicks"), + stream=True, + ) + def stream_cb(n): + yield "a", no_update + yield no_update, no_update # produces no frame + yield no_update, "b" + raise PreventUpdate # ends the stream cleanly + + body = { + "output": "..out.children...out2.children..", + "outputs": [ + {"id": "out", "property": "children"}, + {"id": "out2", "property": "children"}, + ], + "inputs": [{"id": "btn", "property": "n_clicks", "value": 1}], + "changedPropIds": ["btn.n_clicks"], + } + frames = post_stream(app, body) + assert frames[0]["response"] == {"out": {"children": "a"}} + assert frames[1]["response"] == {"out2": {"children": "b"}} + assert frames[2] == {"done": True} + assert len(frames) == 3 + + +def test_stcb012_sync_iter_asyncgen(): + var = contextvars.ContextVar("stcb012") + + async def agen(): + var.set("inside") + for i in range(3): + await asyncio.sleep(0.001) + # The whole generator runs on a single task, so context set + # inside persists across steps. + assert var.get() == "inside" + yield i + + assert list(sync_iter_asyncgen(agen())) == [0, 1, 2] + + +def test_stcb013_sync_iter_asyncgen_error_propagates(): + async def agen(): + yield 1 + raise RuntimeError("kaput") + + gen = sync_iter_asyncgen(agen()) + assert next(gen) == 1 + with pytest.raises(RuntimeError, match="kaput"): + next(gen) + + +def test_stcb014_sync_iter_asyncgen_close_cancels(): + closed = [] + + async def agen(): + try: + for i in range(100): + await asyncio.sleep(0.001) + yield i + finally: + closed.append(True) + + gen = sync_iter_asyncgen(agen()) + assert next(gen) == 0 + gen.close() + # The consumer task is cancelled on a background thread; give it a moment. + for _ in range(100): + if closed: + break + import time + + time.sleep(0.01) + assert closed == [True] diff --git a/tests/websocket/test_ws_stream.py b/tests/websocket/test_ws_stream.py new file mode 100644 index 0000000000..09d641f20d --- /dev/null +++ b/tests/websocket/test_ws_stream.py @@ -0,0 +1,163 @@ +"""WebSocket streaming callback tests. + +Protocol-level tests (FastAPI TestClient, no browser) verifying that +stream=True callbacks emit intermediate callback_response frames with +stream=True followed by a terminal done frame, plus browser tests for the +full renderer round-trip. +""" +import asyncio +import json + +import pytest + +from dash import Dash, Input, Output, Patch, html + + +def _collect_stream_messages(ws): + """Read ws messages, flattening batched arrays, until the terminal frame.""" + out = [] + while True: + parsed = json.loads(ws.receive_text()) + msgs = parsed if isinstance(parsed, list) else [parsed] + for msg in msgs: + if msg.get("type") != "callback_response": + continue + out.append(msg) + payload = msg.get("payload") or {} + if payload.get("done") or not payload.get("stream"): + return out + + +def _make_ws_app(): + from fastapi import FastAPI + + server = FastAPI() + app = Dash(__name__, server=server, websocket_callbacks=True) + app.layout = html.Div([html.Button(id="btn"), html.Div(id="out")]) + return app, server + + +def _callback_request(request_id, output_id="out", prop="children"): + return { + "type": "callback_request", + "requestId": request_id, + "rendererId": "rend1", + "payload": { + "output": f"{output_id}.{prop}", + "outputs": {"id": output_id, "property": prop}, + "inputs": [{"id": "btn", "property": "n_clicks", "value": 1}], + "changedPropIds": ["btn.n_clicks"], + }, + } + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_wsst001_async_stream_frames_over_ws(): + from fastapi.testclient import TestClient + + app, server = _make_ws_app() + + @app.callback(Output("out", "children"), Input("btn", "n_clicks"), stream=True) + async def stream_cb(n): + yield "start" + await asyncio.sleep(0.01) + yield "final" + + app._setup_server() + + client = TestClient(server) + with client.websocket_connect( + "/_dash-ws-callback", headers={"origin": "http://testserver"} + ) as ws: + ws.send_text(json.dumps(_callback_request("r1"))) + msgs = _collect_stream_messages(ws) + + assert [m["requestId"] for m in msgs] == ["r1"] * 3 + assert msgs[0]["payload"]["stream"] is True + assert msgs[0]["payload"]["data"]["response"] == {"out": {"children": "start"}} + assert msgs[1]["payload"]["data"]["response"] == {"out": {"children": "final"}} + assert msgs[2]["payload"] == {"status": "ok", "stream": True, "done": True} + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_wsst002_sync_stream_frames_over_ws(): + from fastapi.testclient import TestClient + + app, server = _make_ws_app() + + @app.callback(Output("out", "children"), Input("btn", "n_clicks"), stream=True) + def stream_cb(n): + yield "s1" + yield "s2" + + app._setup_server() + + client = TestClient(server) + with client.websocket_connect( + "/_dash-ws-callback", headers={"origin": "http://testserver"} + ) as ws: + ws.send_text(json.dumps(_callback_request("r1"))) + msgs = _collect_stream_messages(ws) + + assert msgs[0]["payload"]["data"]["response"] == {"out": {"children": "s1"}} + assert msgs[1]["payload"]["data"]["response"] == {"out": {"children": "s2"}} + assert msgs[2]["payload"] == {"status": "ok", "stream": True, "done": True} + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_wsst003_stream_error_over_ws(): + from fastapi.testclient import TestClient + + app, server = _make_ws_app() + + @app.callback(Output("out", "children"), Input("btn", "n_clicks"), stream=True) + async def stream_cb(n): + yield "one" + raise ValueError("boom") + + app._setup_server() + + client = TestClient(server) + with client.websocket_connect( + "/_dash-ws-callback", headers={"origin": "http://testserver"} + ) as ws: + ws.send_text(json.dumps(_callback_request("r1"))) + msgs = _collect_stream_messages(ws) + + assert msgs[0]["payload"]["data"]["response"] == {"out": {"children": "one"}} + assert msgs[1]["payload"]["status"] == "error" + assert "boom" in msgs[1]["payload"]["message"] + + +def test_wsst004_browser_stream_over_websocket(dash_duo): + """Full round-trip: streamed frames render progressively over WS.""" + app = Dash(__name__, backend="fastapi", websocket_callbacks=True) + app.layout = html.Div( + [ + html.Button("Start", id="btn", n_clicks=0), + html.Div(id="out", children="idle"), + ] + ) + + @app.callback( + Output("out", "children"), + Input("btn", "n_clicks"), + stream=True, + prevent_initial_call=True, + ) + async def stream_cb(n): + yield "streaming" + for token in ["a", "b", "c"]: + await asyncio.sleep(0.2) + patch = Patch() + patch += token + yield patch + + dash_duo.start_server(app) + dash_duo.wait_for_text_to_equal("#out", "idle") + dash_duo.find_element("#btn").click() + # Intermediate frame renders before the stream finishes. + dash_duo.wait_for_text_to_equal("#out", "streaming") + # Patch frames appended exactly once each. + dash_duo.wait_for_text_to_equal("#out", "streamingabc") + assert dash_duo.get_logs() == [] From 65c72b90ff881df75bfe18aa27995273a63411c0 Mon Sep 17 00:00:00 2001 From: philippe Date: Thu, 9 Jul 2026 14:30:47 -0400 Subject: [PATCH 2/3] Fix CI: guard WS stream protocol tests on httpx fastapi.testclient imports starlette.testclient, which requires httpx. Skip the protocol-level stream tests when httpx is not installed and add httpx to the CI requirements so the websocket job actually runs them. --- requirements/ci.txt | 1 + tests/websocket/test_ws_stream.py | 3 +++ 2 files changed, 4 insertions(+) diff --git a/requirements/ci.txt b/requirements/ci.txt index 8e18280d04..178082dc99 100644 --- a/requirements/ci.txt +++ b/requirements/ci.txt @@ -3,6 +3,7 @@ black==22.3.0 flake8==7.0.0 flaky==3.8.1 flask-talisman==1.0.0 +httpx # fastapi.testclient (tests/websocket/test_ws_stream.py) ipython<9.0.0 mimesis<=11.1.0 mock==4.0.3 diff --git a/tests/websocket/test_ws_stream.py b/tests/websocket/test_ws_stream.py index 09d641f20d..f96395a2d8 100644 --- a/tests/websocket/test_ws_stream.py +++ b/tests/websocket/test_ws_stream.py @@ -53,6 +53,7 @@ def _callback_request(request_id, output_id="out", prop="children"): @pytest.mark.filterwarnings("ignore::RuntimeWarning") def test_wsst001_async_stream_frames_over_ws(): + pytest.importorskip("httpx", reason="fastapi.testclient requires httpx") from fastapi.testclient import TestClient app, server = _make_ws_app() @@ -81,6 +82,7 @@ async def stream_cb(n): @pytest.mark.filterwarnings("ignore::RuntimeWarning") def test_wsst002_sync_stream_frames_over_ws(): + pytest.importorskip("httpx", reason="fastapi.testclient requires httpx") from fastapi.testclient import TestClient app, server = _make_ws_app() @@ -106,6 +108,7 @@ def stream_cb(n): @pytest.mark.filterwarnings("ignore::RuntimeWarning") def test_wsst003_stream_error_over_ws(): + pytest.importorskip("httpx", reason="fastapi.testclient requires httpx") from fastapi.testclient import TestClient app, server = _make_ws_app() From e444e726c385bc5fd9e95945d3ecb907e3317ddd Mon Sep 17 00:00:00 2001 From: philippe Date: Mon, 27 Jul 2026 10:02:02 -0400 Subject: [PATCH 3/3] Keep streamed callback responses alive through proxy idle timeouts A stream=True callback that goes quiet between yields sends no bytes, so every proxy in a typical deployment eventually closes the connection on its own idle timeout (nginx's proxy_read_timeout defaults to 60s). The renderer treats a stream that ends without a terminal frame as a dropped connection, so this looked like silent truncation. The NDJSON transports now emit a blank line every stream_keepalive_interval milliseconds (new Dash argument, default 15000, None or 0 disables) that the callback spends between yields. The renderer already skips blank lines, so no client change is needed. The WebSocket transport is unaffected -- it has websocket_heartbeat_interval. The two paths need different mechanics. The async path holds the pending __anext__ across timeouts; asyncio.wait_for would cancel the user generator mid-step every time a keepalive came due. The sync path drives the frame generator on a pump thread, since a blocking next() cannot be interrupted on a timer -- one consequence is that a client disconnect no longer raises GeneratorExit into the user generator at its current yield, which is one more reason sync generators warn at registration. Note this does not address gunicorn's worker timeout, which kills the worker rather than the connection and is only fixable via --worker-class. --- CHANGELOG.md | 2 +- dash/_streaming.py | 138 ++++++++++++++++++++++++++-- dash/backends/_fastapi.py | 6 +- dash/backends/_flask.py | 8 +- dash/backends/_quart.py | 6 +- dash/dash.py | 9 ++ tests/unit/test_stream_callbacks.py | 132 +++++++++++++++++++++++++- 7 files changed, 283 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fc569c8380..519cfe7b4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ This project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] ### Added -- [#]() Streaming callbacks: `@callback(..., stream=True)` marks a generator (or async generator) callback whose yields are pushed to the browser as they are produced. Each yielded value has the same shape as a regular return value and replaces the outputs; yielding `dash.Patch` objects gives incremental updates (e.g. LLM token streaming). Streams ride the WebSocket callback transport when active, otherwise the HTTP response streams NDJSON frames. Works on Flask, Quart, and FastAPI; synchronous generators warn at registration since they occupy a server worker for the whole stream. +- [#3888](https://github.com/plotly/dash/pull/3888/) Streaming callbacks: `@callback(..., stream=True)` marks a generator (or async generator) callback whose yields are pushed to the browser as they are produced. Each yielded value has the same shape as a regular return value and replaces the outputs; yielding `dash.Patch` objects gives incremental updates (e.g. LLM token streaming). Streams ride the WebSocket callback transport when active, otherwise the HTTP response streams NDJSON frames. Works on Flask, Quart, and FastAPI; synchronous generators warn at registration since they occupy a server worker for the whole stream. HTTP streams emit a blank keepalive line every `stream_keepalive_interval` milliseconds (default 15000) that the callback spends between yields, so proxy idle timeouts (nginx's `proxy_read_timeout` defaults to 60s) don't close a stream while the callback is still working; set `stream_keepalive_interval=None` on the app to disable. - [#3646](https://github.com/plotly/dash/pull/3646) Experimental support for React 19. The default is still React 18.3.1; to use React 19 set the environment variable `REACT_VERSION=19.2.4` before running your app, or call `dash._dash_renderer._set_react_version("19.2.4")` inside the app. React 19 has no official UMD builds, so Dash serves the [`umd-react`](https://www.npmjs.com/package/umd-react) package, together with a compatibility shim loaded after react-dom and before any component package. The shim keeps component libraries built against React <=18 (e.g. dash-bootstrap-components, dash-mantine-components) working under React 19: it stubs the removed `ReactCurrentOwner` internals, redirects the legacy element `$$typeof` symbol so pre-bundled React 18 jsx-runtimes produce elements React 19 accepts (error #525), and exposes a global `react/jsx-runtime` (`window.ReactJSXRuntime`) that Dash's own component bundles externalize to. Component library authors adopting this convention should copy the defensive `jsxRuntimeExternal` webpack external from `components/dash-core-components/webpack.config.js` rather than a bare `'ReactJSXRuntime'` string: it falls back to a `React.createElement`-based runtime when the global is missing, so the same build also works on Dash versions older than this release. ### Removed diff --git a/dash/_streaming.py b/dash/_streaming.py index d3a5609dbe..03a15dd9fe 100644 --- a/dash/_streaming.py +++ b/dash/_streaming.py @@ -13,6 +13,15 @@ ``contextvars.ContextVar`` and a sync generator runs in whatever context its consumer drives it from — which, for a streaming HTTP response, is not the request context the callback started in. + +The NDJSON transports also emit keepalives: a blank line every +``keepalive`` seconds the callback spends between yields. Every proxy in a +typical deployment enforces an idle timeout on the response (nginx's +``proxy_read_timeout`` defaults to 60s), and a callback that thinks for +longer than that gets its connection closed mid-stream. A blank line resets +those timers; the renderer skips empty lines, so it costs nothing on the +client. The WebSocket transport has its own heartbeat +(``websocket_heartbeat_interval``) and does not use these helpers. """ import asyncio @@ -32,7 +41,19 @@ # produced (X-Accel-Buffering covers nginx). STREAM_HEADERS = {"X-Accel-Buffering": "no", "Cache-Control": "no-cache"} +# Emitted when the callback is quiet for longer than the keepalive interval. +# The renderer's NDJSON reader skips blank lines. +KEEPALIVE_LINE = "\n" + _SENTINEL = object() +_KEEPALIVE = object() + + +def keepalive_seconds(interval_ms): + """Normalize a configured keepalive interval (ms) to seconds, or None.""" + if not interval_ms or interval_ms <= 0: + return None + return interval_ms / 1000 def to_json(value) -> str: @@ -107,29 +128,130 @@ def _serialize_frame(frame): ) -def ndjson_lines(marker): +def _keepalive_frames(marker, keepalive): + """Yield frames from a sync generator, plus keepalives while it is quiet. + + A blocking ``next()`` cannot be interrupted on a timer, so the frame + generator is driven on a pump thread and this generator waits on a queue + instead. One consequence: a client disconnect no longer raises + ``GeneratorExit`` into the user generator at its current yield — the pump + notices the stop flag once the next frame arrives, and closes it then. The + async path (``async def`` callbacks) cancels at the yield as before, which + is one more reason ``dash._callback`` recommends async for streams. + """ + frames: queue.Queue = queue.Queue(maxsize=1) + stop = threading.Event() + + def put(item): + """Hand one item to the consumer; False if it went away.""" + while not stop.is_set(): + try: + frames.put(item, timeout=0.2) + return True + except queue.Full: + continue + return False + + def pump(): + try: + for frame in iter_stream_frames(marker): + if not put(("item", frame)): + return + put(("end", None)) + except BaseException as err: # pylint: disable=broad-exception-caught + put(("error", err)) + finally: + # Run the user generator's cleanup inside the callback context so + # dash.ctx still resolves in its GeneratorExit/finally handlers. + with contextlib.suppress(Exception): + marker.ctx.run(marker.frames.close) + + thread = threading.Thread(target=pump, daemon=True, name="dash-stream-pump") + thread.start() + try: + while True: + try: + kind, value = frames.get(timeout=keepalive) + except queue.Empty: + yield _KEEPALIVE + continue + if kind == "item": + yield value + elif kind == "error": + raise value + else: + return + finally: + stop.set() + + +async def _akeepalive_frames(frames, keepalive): + """Yield frames from an async iterator, plus keepalives while it is quiet. + + The pending ``__anext__`` is held across timeouts rather than awaited with + ``asyncio.wait_for``, which would cancel the user generator mid-step every + time a keepalive was due. + """ + iterator = frames.__aiter__() + pending = None + try: + while True: + pending = asyncio.ensure_future(iterator.__anext__()) + while True: + done, _ = await asyncio.wait({pending}, timeout=keepalive) + if done: + break + yield _KEEPALIVE + try: + frame = pending.result() + except StopAsyncIteration: + return + finally: + pending = None + yield frame + finally: + # Reached on client disconnect too: cancel the in-flight step so the + # user generator sees CancelledError at its current await. + if pending is not None and not pending.done(): + pending.cancel() + + +def _line(frame): + """Serialize one frame or keepalive; ``(line, fatal)`` as _serialize_frame.""" + if frame is _KEEPALIVE: + return KEEPALIVE_LINE, False + return _serialize_frame(frame) + + +def ndjson_lines(marker, keepalive=None): """Sync NDJSON body for a sync frame generator (Flask/WSGI).""" - for frame in iter_stream_frames(marker): - line, fatal = _serialize_frame(frame) + if keepalive: + frames = _keepalive_frames(marker, keepalive) + else: + frames = iter_stream_frames(marker) + for frame in frames: + line, fatal = _line(frame) yield line if fatal: return -async def andjson_lines(frames): +async def andjson_lines(frames, keepalive=None): """Async NDJSON body over an async iterator of frames.""" + if keepalive: + frames = _akeepalive_frames(frames, keepalive) async for frame in frames: - line, fatal = _serialize_frame(frame) + line, fatal = _line(frame) yield line if fatal: return -def marker_ndjson_aiter(marker): +def marker_ndjson_aiter(marker, keepalive=None): """Async NDJSON body for either flavor of frame generator.""" if marker.is_async: - return andjson_lines(marker.frames) - return andjson_lines(aiter_stream_frames(marker)) + return andjson_lines(marker.frames, keepalive) + return andjson_lines(aiter_stream_frames(marker), keepalive) def sync_iter_asyncgen(agen): diff --git a/dash/backends/_fastapi.py b/dash/backends/_fastapi.py index 669a6e7bda..bce9d6f182 100644 --- a/dash/backends/_fastapi.py +++ b/dash/backends/_fastapi.py @@ -40,6 +40,7 @@ STREAM_HEADERS, STREAM_MIMETYPE, StreamedCallbackResponse, + keepalive_seconds, marker_ndjson_aiter, ) from dash.exceptions import PreventUpdate @@ -575,7 +576,10 @@ async def _dispatch(request: Request): # pylint: disable=unused-argument response_data = await response_data if isinstance(response_data, StreamedCallbackResponse): return StreamingResponse( - marker_ndjson_aiter(response_data), + marker_ndjson_aiter( + response_data, + keepalive_seconds(dash_app._stream_keepalive_interval), + ), media_type=STREAM_MIMETYPE, headers=dict(STREAM_HEADERS), ) diff --git a/dash/backends/_flask.py b/dash/backends/_flask.py index 868c34611f..346f990789 100644 --- a/dash/backends/_flask.py +++ b/dash/backends/_flask.py @@ -35,6 +35,7 @@ STREAM_MIMETYPE, StreamedCallbackResponse, andjson_lines, + keepalive_seconds, ndjson_lines, sync_iter_asyncgen, ) @@ -263,12 +264,15 @@ def serve_callback(self, dash_app: Dash): def _stream_response( marker: StreamedCallbackResponse, with_request_ctx: bool ) -> Response: + keepalive = keepalive_seconds( + dash_app._stream_keepalive_interval # pylint: disable=protected-access + ) if marker.is_async: # Drive the async frame generator on a private event-loop # thread; the response iterator drains it synchronously. - body = sync_iter_asyncgen(andjson_lines(marker.frames)) + body = sync_iter_asyncgen(andjson_lines(marker.frames, keepalive)) else: - body = ndjson_lines(marker) + body = ndjson_lines(marker, keepalive) if with_request_ctx: # Keep flask.request usable while the body is iterated (the # generator runs after the view returns). Only valid on the diff --git a/dash/backends/_quart.py b/dash/backends/_quart.py index 9aee676f00..fd8b8e6c01 100644 --- a/dash/backends/_quart.py +++ b/dash/backends/_quart.py @@ -44,6 +44,7 @@ STREAM_HEADERS, STREAM_MIMETYPE, StreamedCallbackResponse, + keepalive_seconds, marker_ndjson_aiter, ) from dash._utils import parse_version @@ -409,7 +410,10 @@ async def _dispatch(): response_data = await response_data if isinstance(response_data, StreamedCallbackResponse): return Response( # type: ignore[return-value] - marker_ndjson_aiter(response_data), + marker_ndjson_aiter( + response_data, + keepalive_seconds(dash_app._stream_keepalive_interval), + ), content_type=STREAM_MIMETYPE, headers=dict(STREAM_HEADERS), ) diff --git a/dash/dash.py b/dash/dash.py index fd6a8d017d..069577688f 100644 --- a/dash/dash.py +++ b/dash/dash.py @@ -433,6 +433,13 @@ class Dash(ObsoleteChecker): :param csrf_header_name: Name of the HTTP header to send the CSRF token in. Default ``'X-CSRFToken'``. :type csrf_header_name: string + + :param stream_keepalive_interval: How long a ``stream=True`` callback may + go without yielding, in milliseconds, before the response emits a + blank keepalive line. Default 15000. Keeps proxy idle timeouts + (nginx ``proxy_read_timeout`` defaults to 60s) from closing a stream + while the callback is still working. Set to None or 0 to disable. + :type stream_keepalive_interval: int or None """ _plotlyjs_url: str @@ -493,6 +500,7 @@ def __init__( # pylint: disable=too-many-statements, too-many-branches websocket_heartbeat_interval: Optional[int] = 30000, websocket_batch_delay: Optional[float] = 0.005, websocket_max_workers: Optional[int] = 4, + stream_keepalive_interval: Optional[int] = 15000, enable_mcp: Optional[bool] = None, mcp_path: Optional[str] = None, **obsolete, @@ -668,6 +676,7 @@ def __init__( # pylint: disable=too-many-statements, too-many-branches self._websocket_heartbeat_interval = websocket_heartbeat_interval self._websocket_batch_delay = websocket_batch_delay self._websocket_max_workers = websocket_max_workers + self._stream_keepalive_interval = stream_keepalive_interval self.logger = logging.getLogger(__name__) diff --git a/tests/unit/test_stream_callbacks.py b/tests/unit/test_stream_callbacks.py index 117e831373..5b9acc1d59 100644 --- a/tests/unit/test_stream_callbacks.py +++ b/tests/unit/test_stream_callbacks.py @@ -2,12 +2,20 @@ import asyncio import contextvars import json +import time import pytest from dash import Dash, Input, Output, Patch, callback, html, no_update, set_props from dash._callback import GLOBAL_CALLBACK_LIST, GLOBAL_CALLBACK_MAP -from dash._streaming import sync_iter_asyncgen +from dash._streaming import ( + StreamedCallbackResponse, + _keepalive_frames, + andjson_lines, + keepalive_seconds, + marker_ndjson_aiter, + sync_iter_asyncgen, +) from dash.exceptions import ( BackgroundCallbackError, PreventUpdate, @@ -24,13 +32,18 @@ def make_body(output_id, prop, input_id="btn"): } -def post_stream(app, body): - """POST a callback request and return the parsed NDJSON frames.""" +def post_stream_raw(app, body): + """POST a callback request and return the raw NDJSON body.""" client = app.server.test_client() resp = client.post("/_dash-update-component", json=body) assert resp.status_code == 200 assert resp.headers.get("Content-Type") == "application/x-ndjson" - data = resp.get_data(as_text=True) + return resp.get_data(as_text=True) + + +def post_stream(app, body): + """POST a callback request and return the parsed NDJSON frames.""" + data = post_stream_raw(app, body) return [json.loads(line) for line in data.splitlines() if line.strip()] @@ -277,7 +290,116 @@ async def agen(): for _ in range(100): if closed: break - import time + time.sleep(0.01) + assert closed == [True] + + +def test_stcb015_keepalive_seconds_normalization(): + assert keepalive_seconds(15000) == 15.0 + assert keepalive_seconds(None) is None + assert keepalive_seconds(0) is None + assert keepalive_seconds(-1) is None + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stcb016_flask_keepalive_between_slow_yields(): + app = Dash(__name__, stream_keepalive_interval=50) + app.layout = html.Div([html.Button(id="btn"), html.Div(id="out")]) + + @app.callback(Output("out", "children"), Input("btn", "n_clicks"), stream=True) + def stream_cb(n): + time.sleep(0.3) + yield "start" + time.sleep(0.3) + yield "final" + + raw = post_stream_raw(app, make_body("out", "children")) + # Blank keepalive lines while the callback is between yields. + assert len([line for line in raw.splitlines() if not line.strip()]) >= 2 + # The frames themselves are unaffected. + frames = [json.loads(line) for line in raw.splitlines() if line.strip()] + assert frames[0]["response"] == {"out": {"children": "start"}} + assert frames[1]["response"] == {"out": {"children": "final"}} + assert frames[2] == {"done": True} + + +@pytest.mark.filterwarnings("ignore::RuntimeWarning") +def test_stcb017_flask_keepalive_disabled(): + app = Dash(__name__, stream_keepalive_interval=None) + app.layout = html.Div([html.Button(id="btn"), html.Div(id="out")]) + + @app.callback(Output("out", "children"), Input("btn", "n_clicks"), stream=True) + def stream_cb(n): + time.sleep(0.2) + yield "only" + + raw = post_stream_raw(app, make_body("out", "children")) + assert [line for line in raw.splitlines() if not line.strip()] == [] + + +def test_stcb018_async_keepalive_does_not_cancel_source(): + async def agen(): + await asyncio.sleep(0.3) + yield {"multi": True} + await asyncio.sleep(0.3) + yield {"done": True} + + async def collect(): + return [line async for line in andjson_lines(agen(), keepalive=0.05)] + + lines = asyncio.run(collect()) + assert lines.count("\n") >= 2 + # Holding the pending __anext__ across keepalives means both frames still + # arrive; a bare wait_for would have cancelled the generator mid-step. + assert [json.loads(line) for line in lines if line.strip()] == [ + {"multi": True}, + {"done": True}, + ] + + +def test_stcb020_async_keepalive_over_sync_generator(): + """marker_ndjson_aiter with is_async=False: sync generator, ASGI backend.""" + + def frames(): + time.sleep(0.3) + yield {"multi": True} + yield {"done": True} + + marker = StreamedCallbackResponse( + frames(), is_async=False, ctx=contextvars.copy_context() + ) + + async def collect(): + return [line async for line in marker_ndjson_aiter(marker, keepalive=0.05)] + + lines = asyncio.run(collect()) + assert lines.count("\n") >= 2 + assert [json.loads(line) for line in lines if line.strip()] == [ + {"multi": True}, + {"done": True}, + ] + + +def test_stcb019_keepalive_frames_closes_generator_when_consumer_leaves(): + closed = [] + + def frames(): + try: + while True: + yield {"multi": True} + finally: + closed.append(True) + + marker = StreamedCallbackResponse( + frames(), is_async=False, ctx=contextvars.copy_context() + ) + gen = _keepalive_frames(marker, 0.05) + assert next(gen) == {"multi": True} + gen.close() + # The pump thread owns the generator, so cleanup happens once it notices + # the stop flag rather than at the consumer's close(). + for _ in range(200): + if closed: + break time.sleep(0.01) assert closed == [True]