如何防止 Dash 应用程序在 long_callback、Python3 期间受到用户影响?

How to prevent a Dash app influenced by user during a long_callback, Python3?

我做了一个app:用Dash做浏览器的gui,后台计算纯Python3。

由于根据用户的研究案例,计算可能需要几秒到几小时甚至几天,所以我使用了来自 Dash 的 long_callback 的装饰器。 GUI 遵循简单的用户逻辑:

  1. 当点击运行按钮时,该按钮被禁用,不仅可以告知用户该应用正在计算,还可以防止用户在该应用仍在计算时再次点击该按钮计算。
  2. 当应用程序仍在进行计算时,用户应该能够使用应用程序的其他部分,但 运行 按钮除外。后台正在进行的计算应该不会受到影响。

我做了一个简化的代码来演示我的问题。

  1. 首先,我输入一个任意值,例如条目中有 10 个。
  2. 然后,我单击 运行 按钮。该按钮已禁用,应用程序 运行 符合预期。
  3. 在 运行 完成之前(即在 运行 按钮启用之前),我将条目从 10 更改为 20,输出消息显示数字 20 而不是10。这怎么可能?单击 运行 按钮后,GUI 上的任何进一步操作都不应影响已经开始的调用。你能告诉我如何实施吗?谢谢。
import time
import dash
from dash import html, dcc
from dash.long_callback import DiskcacheLongCallbackManager
from dash.dependencies import Input, Output, State


# Diskcache
import diskcache

cache = diskcache.Cache("./cache")
long_callback_manager = DiskcacheLongCallbackManager(cache)

app = dash.Dash(__name__, long_callback_manager=long_callback_manager)

app.layout = html.Div(
    [
        html.Div([html.P(id="paragraph_id", children=["Button not clicked"])]),
        html.Button(id="button_id", children="Run Job!"),
        dcc.Input(id='entry_id')
    ]
)


@app.long_callback(
    output=Output("paragraph_id", "children"),
    inputs=dict(
        n_clicks=Input("button_id", "n_clicks"),
        entry_text=State("entry_id", "value"),
    ),
    running=[
        (Output("button_id", "disabled"), True, False),
    ],
    prevent_initial_call=True,
)
def callback(n_clicks, entry_text):
    if not n_clicks:
        raise dash.exceptions.PreventUpdate

    time.sleep(3.0)  # Here 3 seconds is just an example. My actual code can run days.
    return [f"Clicked {n_clicks} times, entered {entry_text}"]


if __name__ == "__main__":
    app.run_server(debug=True)

如果您希望 State 不影响现有的回调执行,请在执行期间将该组件的禁用设置为 True,就像您在回调计算期间禁用按钮的方式一样:

@app.long_callback(
    output=Output("paragraph_id", "children"),
    inputs=dict(
        n_clicks=Input("button_id", "n_clicks"),
        entry_text=State("entry_id", "value"),
    ),
    running=[
        (Output("button_id", "disabled"), True, False),
        (Output("entry_id", "disabled"), True, False),
    ],
    prevent_initial_call=True,
)
def callback(n_clicks, entry_text):
    if not n_clicks:
        raise dash.exceptions.PreventUpdate

    time.sleep(3.0)  # Here 3 seconds is just an example. My actual code can run days.
    return [f"Clicked {n_clicks} times, entered {entry_text}"]

TLDR:您可以通过添加一个 Store 元素来解决这个问题,该元素在调度计算之前缓存当前值,

import time
import dash
import diskcache

from dash import html, dcc
from dash.long_callback import DiskcacheLongCallbackManager
from dash.dependencies import Input, Output, State


# Diskcache
cache = diskcache.Cache("./cache")
long_callback_manager = DiskcacheLongCallbackManager(cache)

app = dash.Dash(__name__, long_callback_manager=long_callback_manager)

app.layout = html.Div(
    [
        html.Div([html.P(id="paragraph_id", children=["Button not clicked"])]),
        html.Button(id="button_id", children="Run Job!"),
        dcc.Input(id="entry_id"),
        dcc.Store(id="input_cache"),
    ]
)


@app.callback(
    Output("input_cache", "data"), Input("button_id", "n_clicks"), State("entry_id", "value")
)
def cache_job_inputs(n_clicks, entry_text):
    if not n_clicks:
        raise dash.exceptions.PreventUpdate
    return dict(n_clicks=n_clicks, entry_text=entry_text)


@app.long_callback(
    output=Output("paragraph_id", "children"),
    inputs=dict(
        data=Input("input_cache", "data"),
    ),
    running=[
        (Output("button_id", "disabled"), True, False),
    ],
    prevent_initial_call=True,
)
def callback(data):
    time.sleep(3.0)  # Here 3 seconds is just an example. My actual code can run days.
    return [f"Clicked {data['n_clicks']} times, entered {data['entry_text']}"]


if __name__ == "__main__":
    app.run_server(debug=True)

通过 dash.callback_context.triggered 属性 检查回调调用,我注意到 _long_callback_interval_1.n_intervals 的常规调用。 entry_id 的新值在这些调用期间被捕获,这对我来说似乎是一个错误。