diff --git a/.ai/ARCHITECTURE.md b/.ai/ARCHITECTURE.md index d90f670f9b..7f40ee47a9 100644 --- a/.ai/ARCHITECTURE.md +++ b/.ai/ARCHITECTURE.md @@ -1057,6 +1057,109 @@ 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 + +A callback defined as a generator function (or async generator function) +streams: its yields are pushed to the browser as they are produced — for LLM +token streaming, progress feeds, and long computations. There is no opt-in +keyword; `dash._callback.register_callback` infers it from the decorated +function (`inspect.isgeneratorfunction` / `isasyncgenfunction`) and registers +the streaming wrapper instead of the regular one. + +```python +import asyncio +from dash import callback, Output, Input, Patch + +@callback( + Output('log', 'children'), + Input('btn', 'n_clicks'), + 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`, `mcp_enabled` and `api_endpoint` + (validated at registration, when the function is inspected). Clientside + callbacks cannot stream at all. +- `callback_map[callback_id]['stream']` records the inferred flag server-side; + it is not part of the callback spec sent to the client, which detects a + stream from the response instead (NDJSON content type / `stream` frames). + +### 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 + +- Streaming is inferred from the decorated function, so another decorator + between `@callback` and the generator hides it: if that decorator returns a + plain function, Dash registers a regular callback and the returned generator + object fails to serialize (`InvalidCallbackReturnValue: type generator`). +- 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 5c4cff63c9..2a601b9791 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ This project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] ### Added +- [#3888](https://github.com/plotly/dash/pull/3888/) Streaming callbacks: a callback defined as a generator (or async generator) function streams — its yields are pushed to the browser as they are produced, no keyword needed. 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/_callback.py b/dash/_callback.py index 61908942ef..1a2b9329e2 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 . import _callback_signing @@ -41,6 +44,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 @@ -60,6 +64,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] = [] @@ -109,6 +115,14 @@ def callback( not to fire when its outputs are first added to the page. Defaults to `False` and unlike `app.callback` is not configurable at the app level. + Decorating a generator function (or async generator function) registers a + streaming callback: 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). Streaming callbacks cannot be combined with + `background=True`, `mcp_enabled=True` or `api_endpoint`. + :Keyword Arguments: :param background: Mark the callback as a background callback to execute in a manager for @@ -256,6 +270,25 @@ def callback( ) +def _validate_stream_callback(callback_id, background, kwargs): + """Reject options a streaming (generator) callback cannot be combined with.""" + if background is not None: + raise BackgroundCallbackError( + f"Streaming callback '{callback_id}' cannot be combined with " + "background=True: background callbacks return a single result." + ) + if kwargs.get("mcp_enabled"): + raise StreamCallbackError( + f"Streaming callback '{callback_id}' cannot be combined with " + "mcp_enabled=True: MCP tools expect a single JSON result." + ) + if kwargs.get("api_endpoint"): + raise StreamCallbackError( + f"Streaming callback '{callback_id}' cannot be combined with " + "api_endpoint: API endpoints expect a single JSON result." + ) + + def validate_background_inputs(deps): for dep in deps: if dep.has_wildcard(): @@ -347,6 +380,9 @@ def insert_callback( "allow_dynamic_callbacks": dynamic_creator, "no_output": no_output, "websocket": websocket, + # Flipped to True by register_callback when the decorated function + # turns out to be a generator (streaming callback). + "stream": False, "mcp_enabled": mcp_enabled, "mcp_expose_docstring": mcp_expose_docstring, } @@ -777,6 +813,14 @@ def register_callback( # pylint: disable=too-many-locals def wrap_func(func): + # A generator (or async generator) callback streams its yields; that is + # inferred from the function itself, there is no opt-in keyword. + is_gen_func = inspect.isgeneratorfunction(func) + is_async_gen_func = inspect.isasyncgenfunction(func) + is_stream = is_gen_func or is_async_gen_func + if is_stream: + _validate_stream_callback(callback_id, background, _kwargs) + if _kwargs.get("api_endpoint"): api_endpoint = _kwargs.get("api_endpoint") GLOBAL_API_PATHS[api_endpoint] = func @@ -935,7 +979,205 @@ 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 streaming 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 streaming 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) + + if is_stream: + callback_map[callback_id]["stream"] = True + 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"Streaming 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 inspect.iscoroutinefunction(func): callback_map[callback_id]["callback"] = async_add_context else: # A persistent, no-output callback streams via set_props and typically diff --git a/dash/_streaming.py b/dash/_streaming.py new file mode 100644 index 0000000000..bf9ec39a49 --- /dev/null +++ b/dash/_streaming.py @@ -0,0 +1,311 @@ +"""Transport helpers for streaming callbacks (generator callbacks). + +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. + +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 +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"} + +# 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: + return cast(str, _to_json(value)) + + +class StreamedCallbackResponse: # pylint: disable=too-few-public-methods + """Marker returned by streaming 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 _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).""" + 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, 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 = _line(frame) + yield line + if fatal: + return + + +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, keepalive) + return andjson_lines(aiter_stream_frames(marker), keepalive) + + +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..bce9d6f182 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,13 @@ from dash.fingerprint import check_fingerprint from dash import _validate, get_app +from dash._streaming import ( + STREAM_HEADERS, + STREAM_MIMETYPE, + StreamedCallbackResponse, + keepalive_seconds, + marker_ndjson_aiter, +) from dash.exceptions import PreventUpdate from .base_server import ( BaseDashServer, @@ -48,6 +55,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 +574,15 @@ 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, + keepalive_seconds(dash_app._stream_keepalive_interval), + ), + media_type=STREAM_MIMETYPE, + headers=dict(STREAM_HEADERS), + ) return cb_ctx.dash_response.set_response(data=response_data) return _dispatch @@ -812,6 +829,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 +843,7 @@ async def websocket_handler(websocket: WebSocket): payload, ws_cb, FastAPIResponseAdapter(), + stream_emitter, ) ) task.add_done_callback(done_handler) @@ -832,6 +856,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 43f24634a2..346f990789 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,15 @@ 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, + keepalive_seconds, + ndjson_lines, + sync_iter_asyncgen, +) from dash._utils import parse_version from .base_server import BaseDashServer, RequestAdapter, ResponseAdapter @@ -251,6 +261,33 @@ 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: + 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, keepalive)) + else: + 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 + # 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 +299,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 +322,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) # Preserve the view function's identity as `dash.dash.dispatch` so that diff --git a/dash/backends/_quart.py b/dash/backends/_quart.py index 9487cac7d8..fd8b8e6c01 100644 --- a/dash/backends/_quart.py +++ b/dash/backends/_quart.py @@ -40,6 +40,13 @@ from dash.exceptions import PreventUpdate, InvalidResourceError from dash.fingerprint import check_fingerprint +from dash._streaming import ( + STREAM_HEADERS, + STREAM_MIMETYPE, + StreamedCallbackResponse, + keepalive_seconds, + marker_ndjson_aiter, +) from dash._utils import parse_version from dash import _validate from .base_server import ( @@ -53,6 +60,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 +408,15 @@ 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, + keepalive_seconds(dash_app._stream_keepalive_interval), + ), + content_type=STREAM_MIMETYPE, + headers=dict(STREAM_HEADERS), + ) return cb_ctx.dash_response.set_response(data=response_data) # type: ignore[arg-type] # Preserve the view function's identity as `dash.dash.dispatch` so that @@ -652,6 +669,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( @@ -660,6 +683,7 @@ async def websocket_handler(): # pylint: disable=too-many-branches payload, ws_cb, QuartResponseAdapter(), + stream_emitter, ) ) task.add_done_callback(done_handler) @@ -672,6 +696,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 6dec01996c..61fc53d0a5 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 streaming 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, @@ -614,7 +644,92 @@ function handleServerside( } }; + // Streaming 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; @@ -726,7 +841,16 @@ async function handleWebsocketCallback( // Ensure WebSocket connection is established await workerClient.ensureConnected(config); - const response = await workerClient.sendCallback(payload); + // Streaming 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) { @@ -760,6 +884,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..bd8a6a1493 100644 --- a/dash/dash-renderer/src/types/callbacks.ts +++ b/dash/dash-renderer/src/types/callbacks.ts @@ -109,6 +109,9 @@ export type CallbackResponseData = { cancel?: ICallbackProperty[]; dist?: any; sideUpdate?: any; + // Streaming 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..b09433d85c 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 streaming 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 streaming 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 + * streaming 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/dash.py b/dash/dash.py index fd6a8d017d..8f5e4ee1eb 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 streaming 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/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..175a1f61ca 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]] + # Streaming callbacks: terminal frame marker and mid-stream error. + done: NotRequired[bool] + error: NotRequired[Dict[str, str]] 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/integration/callbacks/test_stream_callbacks.py b/tests/integration/callbacks/test_stream_callbacks.py new file mode 100644 index 0000000000..12bc60b642 --- /dev/null +++ b/tests/integration/callbacks/test_stream_callbacks.py @@ -0,0 +1,211 @@ +"""Browser integration tests for streaming 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"), + 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"), + 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"), + 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"), + 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"), + 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"), + 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..702ea13641 --- /dev/null +++ b/tests/unit/test_stream_callbacks.py @@ -0,0 +1,385 @@ +"""Unit tests for streaming (generator) callbacks - no browser required.""" +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 ( + StreamedCallbackResponse, + _keepalive_frames, + andjson_lines, + keepalive_seconds, + marker_ndjson_aiter, + 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_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" + 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()] + + +def test_stcb001_non_generator_is_not_streamed(): + @callback(Output("stcb001", "children"), Input("in", "value")) + def not_a_generator(value): + return value + + assert GLOBAL_CALLBACK_MAP["stcb001.children"]["stream"] is False + + +def test_stcb002_generator_streams_without_a_keyword(): + @callback(Output("stcb002", "children"), Input("in", "value")) + async def a_generator(value): + yield value + + assert GLOBAL_CALLBACK_MAP["stcb002.children"]["stream"] is True + + +def test_stcb003_stream_incompatible_kwargs(): + with pytest.raises(BackgroundCallbackError): + + @callback( + Output("stcb003a", "children"), + Input("in", "value"), + background=True, + ) + def bg(value): + yield value + + with pytest.raises(StreamCallbackError, match="mcp_enabled"): + + @callback( + Output("stcb003b", "children"), + Input("in", "value"), + mcp_enabled=True, + ) + def mcp(value): + yield value + + with pytest.raises(StreamCallbackError, match="api_endpoint"): + + @callback( + Output("stcb003c", "children"), + Input("in", "value"), + api_endpoint="/stream", + ) + def api(value): + yield value + + +def test_stcb005_sync_generator_warns(): + with pytest.warns(RuntimeWarning, match="synchronous generator"): + + @callback(Output("stcb005", "children"), Input("in", "value")) + def sync_gen(value): + yield value + + +def test_stcb006_async_generator_no_warning(recwarn): + @callback(Output("stcb006", "children"), Input("in", "value")) + async def async_gen(value): + yield value + + assert not [w for w in recwarn.list if issubclass(w.category, RuntimeWarning)] + + +def test_stcb007_stream_wrapper_registered(): + @callback(Output("stcb007", "children"), Input("in", "value")) + async def async_gen(value): + yield value + + assert GLOBAL_CALLBACK_MAP["stcb007.children"]["stream"] is True + + # The flag is server-side only: the client detects a stream from the + # response (NDJSON content type / stream frames), not the callback spec. + spec = [s for s in GLOBAL_CALLBACK_LIST if s["output"] == "stcb007.children"][-1] + assert "stream" not in spec + + +@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")) + 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")) + 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"), + 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"), + ) + 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 + 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")) + 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")) + 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] diff --git a/tests/websocket/test_ws_stream.py b/tests/websocket/test_ws_stream.py new file mode 100644 index 0000000000..12b231793c --- /dev/null +++ b/tests/websocket/test_ws_stream.py @@ -0,0 +1,165 @@ +"""WebSocket streaming callback tests. + +Protocol-level tests (FastAPI TestClient, no browser) verifying that +streaming 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(): + pytest.importorskip("httpx", reason="fastapi.testclient requires httpx") + from fastapi.testclient import TestClient + + app, server = _make_ws_app() + + @app.callback(Output("out", "children"), Input("btn", "n_clicks")) + 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(): + pytest.importorskip("httpx", reason="fastapi.testclient requires httpx") + from fastapi.testclient import TestClient + + app, server = _make_ws_app() + + @app.callback(Output("out", "children"), Input("btn", "n_clicks")) + 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(): + pytest.importorskip("httpx", reason="fastapi.testclient requires httpx") + from fastapi.testclient import TestClient + + app, server = _make_ws_app() + + @app.callback(Output("out", "children"), Input("btn", "n_clicks")) + 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"), + 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() == []