如何在Dash中结合tqdm实现进度条可视化？

我有以下用于查询数据的代码：

def get_symbol_price(symbol):
    response = client.fetch_prices(symbol)
    df = pd.DataFrame(response)
    df['symbol'] = symbol
    return df

def get_prices(symbols):
    with ThreadPoolExecutor(4) as executor:
        results = list(tqdm(executor.map(get_symbol_klines, symbols), total=len(symbols)))

    df = pd.concat([result for result in results])
    return df

我的Dash应用代码如下：

app.layout = dbc.Container(
    [
        dbc.Button(id='button'),
        dbc.Progress(id="progress", value=0, striped=True, animated=True),
        dcc.Graph(id='graphs'),
    ],
    fluid=True,
)

@app.callback(
    Output("graphs", "figure"),
    Output("progress", "value"),
    Input("button", "n_clicks"),
)
def update_session(n, interval, data):
    if n is None:
        raise PreventUpdate

    prices = get_prices(symbols)

    fig = go.Figure()

    for symbol, value in prices.groupby('symbol'):
        fig.add_scatter(
            x=value.index,
            y=value,
            name=symbol
        )

    fig.update_yaxes(fixedrange=False, automargin=True)
    fig.update_layout(xaxis_rangeslider_visible=True)

    return fig

由于需要查询大量Symbol数据，耗时可达1分钟，请问如何将tqdm的进度显示在Dash的进度条组件中？

解决方案 #

要实现进度同步，核心是让后台任务能实时把进度传递给Dash前端，需结合dcc.Interval组件做进度轮询，加上进度存储机制，具体步骤如下：

1. 调整布局，添加进度轮询组件 #

在布局中加入dcc.Interval，用于定时拉取最新进度：

app.layout = dbc.Container(
    [
        dbc.Button(id='button'),
        dbc.Progress(id="progress", value=0, striped=True, animated=True),
        dcc.Graph(id='graphs'),
        # 每500ms查询一次进度，初始状态为禁用
        dcc.Interval(id='progress-interval', interval=500, disabled=True)
    ],
    fluid=True,
)

2. 搭建进度存储机制 #

用全局字典存储进度（多用户场景下建议结合session ID区分，避免进度混淆）：

# 全局存储进度和任务结果，key用session ID标识
task_store = {}

3. 修改数据查询函数，注入进度更新逻辑 #

改造get_prices，让tqdm的进度实时写入存储：

def get_prices(symbols, session_id):
    total_symbols = len(symbols)
    # 初始化进度为0
    task_store[session_id] = {'progress': 0, 'result': None}

    def update_progress(_):
        # 每次完成一个查询，进度+1
        task_store[session_id]['progress'] += 1

    with ThreadPoolExecutor(4) as executor:
        # 注意：原代码中get_symbol_klines应为笔误，替换为get_symbol_price
        results = list(tqdm(
            executor.map(get_symbol_price, symbols),
            total=total_symbols,
            postfix=update_progress
        ))

    df = pd.concat([result for result in results])
    # 任务完成后存储结果，进度设为总数
    task_store[session_id]['progress'] = total_symbols
    task_store[session_id]['result'] = df
    return df

4. 拆分回调，实现异步任务+进度同步 #

Dash默认同步回调会阻塞前端，需用后台任务处理耗时操作，这里推荐用Dash官方的background_callback（Dash 2.7+支持）：

配置后台回调管理器

from dash.long_callback import DiskcacheLongCallbackManager
import diskcache

# 用diskcache存储后台任务状态
cache = diskcache.Cache("./dash_cache")
long_callback_manager = DiskcacheLongCallbackManager(cache)

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

触发后台任务的回调

@app.callback(
    Output('progress-interval', 'disabled'),
    Input('button', 'n_clicks'),
    State('dash-session-id', 'data'),
    prevent_initial_call=True
)
def start_data_query(n_clicks, session_id):
    # 启动后台数据查询任务
    generate_graph.delay(session_id)
    # 开启进度轮询
    return False

后台数据查询与图表生成

@app.long_callback(
    Output('graphs', 'figure'),
    Input('button', 'n_clicks'),
    State('dash-session-id', 'data'),
    prevent_initial_call=True
)
def generate_graph(session_id):
    # 执行数据查询
    prices = get_prices(symbols, session_id)
    
    # 生成图表
    fig = go.Figure()
    for symbol, value in prices.groupby('symbol'):
        fig.add_scatter(
            x=value.index,
            y=value.iloc[:, 0],  # 取DataFrame第一列作为价格数据
            name=symbol
        )
    fig.update_yaxes(fixedrange=False, automargin=True)
    fig.update_layout(xaxis_rangeslider_visible=True)
    
    # 任务完成后清理存储
    del task_store[session_id]
    return fig

进度条更新回调

@app.callback(
    Output('progress', 'value'),
    Input('progress-interval', 'n_intervals'),
    State('dash-session-id', 'data'),
    prevent_initial_call=True
)
def update_progress_bar(n_intervals, session_id):
    task_data = task_store.get(session_id, {'progress': 0})
    total = len(symbols)
    # 计算进度百分比
    return int((task_data['progress'] / total) * 100) if total > 0 else 0

注意事项 #

• 原代码中get_prices里的get_symbol_klines为笔误，需替换为get_symbol_price
• 多用户场景必须用dash-session-id作为存储key，避免进度混淆
• 生产环境建议用Redis等分布式缓存替代全局字典，防止进程重启丢失数据

内容的提问来源于stack exchange，提问作者Nicolas Rey