在Dash应用中使用websocket型回调#首先我们需要在应用实例化对应的dash.Dash()中设置参数backendfastapi这是令Dash应用底层完美开启websocket通信的必要前提。app dash.Dash(__name__, backendfastapi)在此基础上只需要在对应常规服务端回调函数的callback()中设置websocketTrue该服务端回调函数底层网络通信就会从默认的http传输切换为高性能websocket传输下面是一个简单的例子import dash from dash import html import feffery_antd_components as fac from feffery_dash_utils.style_utils import style from dash.dependencies import Input, Output app dash.Dash(__name__, backendfastapi) app.layout html.Div( [ fac.AntdSpace( [ fac.AntdInput(idinput, placeholder请输入内容), fac.AntdText(idoutput), ] ) ], stylestyle(padding50), ) app.callback(Output(output, children), Input(input, value), websocketTrue) def basic_demo(input_text): return input_text if __name__ __main__: app.run(debugTrue)切换为websocket通信模式后除了响应延时上的显著提升⚡以外更重要的是借助websocket双向通信的特点新版本Dash中可以实现在回调函数里随时随地与前端界面进行数据获取更新下面我们简单举几个例子2.2.1 在websocket回调中实时更新应用#基于新型websocket回调我们可以在回调函数中配合set_props()实时更新相关组件属性譬如下面的实时进度条更新示例import dash import time from dash import html, set_props import feffery_antd_components as fac from feffery_dash_utils.style_utils import style from dash.dependencies import Input app dash.Dash(__name__, backendfastapi) app.layout html.Div( [ fac.AntdSpace( [ fac.AntdButton( 执行任务, idexecute, loadingChildren执行中, typeprimary, autoSpinTrue, ), fac.AntdProgress(idprogress, percent0, stylestyle(width200)), ] ) ], stylestyle(padding50), ) app.callback(Input(execute, nClicks), websocketTrue, prevent_initial_callTrue) def update_progress(nClicks): for i in range(10): time.sleep(0.25) # 模拟计算耗时 set_props(progress, {percent: (i 1) * 10}) set_props(execute, {loading: False}) if __name__ __main__: app.run(debugTrue)2.2.2 在websocket回调中实时获取组件属性#更高级的是基于新型websocket回调函数我们还可以在回调函数中实时获取任意指定id组件对应的任意属性值在异步回调函数中使用譬如下面的例子我们没有事先在回调函数编排中声明两个输入框角色但基于get_prop()就可以在回调函数运行的过程中即时获取想要的属性值import dash from dash import html import feffery_antd_components as fac from feffery_dash_utils.style_utils import style from dash.dependencies import Input app dash.Dash(__name__, backendfastapi) app.layout html.Div( [ fac.AntdSpace( [ fac.AntdButton(触发回调, idtrigger, typeprimary), fac.AntdInput(idinput1), fac.AntdInput(idinput2), ] ) ], stylestyle(padding50), ) app.callback(Input(trigger, nClicks), websocketTrue, prevent_initial_callTrue) async def get_prop_demo(nClicks): ws dash.ctx.websocket value1 await ws.get_prop(input1, value) value2 await ws.get_prop(input2, value) print(value1, value2)
