Plotly/Dash中@callback无法全局修改变量的问题及解决方法

我正在使用Plotly/Dash制作业务仪表盘，希望每5分钟从数据库重新加载数据，因此设置了dcc.Interval()来触发load_data()函数调用。但回调函数更新全局字典data后，下次回调时变量又回到服务器启动时的初始值，尝试过global关键字但Dash官方不推荐该用法，求可行解决方法。

原代码 #

import pandas as pd
import numpy as np
import datetime as dt
from dash import Dash, dcc, html, Input, Output

def load_data():
    N = 100
    df = pd.DataFrame({
        'category': (
            (['apples'] * 5 * N) +
            (['oranges'] * 10 * N) +
            (['figs'] * 20 * N) +
            (['pineapples'] * 15 * N)
        )
    })
    df['x'] = np.random.randn(len(df['category']))
    df['y'] = np.random.randn(len(df['category']))
    load_data_time = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return df, load_data_time

data = dict()
data['df'], data['last_trigger_time'] = load_data()

app = Dash(__name__)
app.layout = html.Div([
    html.Div(id='update-data-time'),
    dcc.Interval(id='interval-component', interval=60*1000, n_intervals=0) # 每分钟触发一次
])

@callback(Output('update-data-time', 'children'),
          Input('interval-component', 'n_intervals'))
def update_data(n):
    current_time = dt.datetime.now()
    print(f"updata data time is {data['last_trigger_time']}")
    if (current_time.minute % 5 == 0):
        data['df'], data['last_trigger_time'] = load_data()
        print("Callback triggered at every 5 minutes")
    else:
        print(f"Waiting for the next 5 minutes...")
    print(f"Last trigger time is {data['last_trigger_time']}")
    return f"last data update time is {data['last_trigger_time']}"

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

原输出 #

updata data time is 2023-07-07 14:38:55
Waiting for the next 5 minutes...
Last trigger time is 2023-07-07 14:38:55
127.0.0.1 - - [07/Jul/2023 14:39:10] "POST /_dash-update-component HTTP/1.1" 200 - 
updata data time is 2023-07-07 14:38:55
Callback triggered at every 5 minutes
Last trigger time is 2023-07-07 14:40:00
127.0.0.1 - - [07/Jul/2023 14:40:00] "POST /_dash-update-component HTTP/1.1" 200 - # trigger, update
updata data time is 2023-07-07 14:38:55
Waiting for the next 5 minutes...
Last trigger time is 2023-07-07 14:38:55 # fail, expect 14:40:00
127.0.0.1 - - [07/Jul/2023 14:41:14] "POST /_dash-update-component HTTP/1.1" 200 - 

可行解决方案 #

1. 使用dcc.Store组件（官方推荐） #

Dash的dcc.Store是专门用于在客户端存储状态的组件，符合Dash的无状态设计理念，能安全地在回调间共享数据。需要将数据和更新时间序列化为JSON格式存储，回调通过State读取当前值，Output更新存储内容和显示文本。

修改后的代码：

import pandas as pd
import numpy as np
import datetime as dt
from dash import Dash, dcc, html, Input, Output, State
import json

def load_data():
    N = 100
    df = pd.DataFrame({
        'category': (
            (['apples'] * 5 * N) +
            (['oranges'] * 10 * N) +
            (['figs'] * 20 * N) +
            (['pineapples'] * 15 * N)
        )
    })
    df['x'] = np.random.randn(len(df['category']))
    df['y'] = np.random.randn(len(df['category']))
    load_data_time = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    # 将DataFrame转为JSON格式，方便存储到dcc.Store
    df_json = df.to_json(orient='split')
    return df_json, load_data_time

app = Dash(__name__)
# 初始化存储数据
initial_df, initial_time = load_data()
app.layout = html.Div([
    html.Div(id='update-data-time'),
    # 存储数据和更新时间的Store组件
    dcc.Store(
        id='data-store',
        data={
            'df': initial_df,
            'last_trigger_time': initial_time
        }
    ),
    dcc.Interval(id='interval-component', interval=60*1000, n_intervals=0) # 每分钟触发一次
])

@callback(
    [Output('update-data-time', 'children'),
     Output('data-store', 'data')],
    Input('interval-component', 'n_intervals'),
    State('data-store', 'data')
)
def update_data(n, stored_data):
    current_time = dt.datetime.now()
    # 从存储中读取当前状态
    last_trigger_time = stored_data['last_trigger_time']
    print(f"updata data time is {last_trigger_time}")
    
    if current_time.minute % 5 == 0:
        # 加载新数据并更新存储内容
        new_df, new_time = load_data()
        stored_data['df'] = new_df
        stored_data['last_trigger_time'] = new_time
        print("Callback triggered at every 5 minutes")
        print(f"Last trigger time is {new_time}")
        return f"last data update time is {new_time}", stored_data
    else:
        print(f"Waiting for the next 5 minutes...")
        print(f"Last trigger time is {last_trigger_time}")
        return f"last data update time is {last_trigger_time}", stored_data

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

2. 使用Flask缓存（适合数据缓存场景） #

如果需要将数据缓存（比如避免频繁查询数据库），可以使用Flask-Caching扩展，将数据存在内存、Redis或其他缓存介质中，跨回调共享。

示例步骤：

• 安装依赖：pip install flask-caching
• 修改代码：

from flask_caching import Cache

app = Dash(__name__)
cache = Cache(app.server, config={
    'CACHE_TYPE': 'SimpleCache', # 单进程内存缓存，多进程请用Redis等
    'CACHE_DEFAULT_TIMEOUT': 300 # 默认5分钟过期
})

def load_data():
    # 原加载数据逻辑
    ...

# 初始化缓存
cache.set('data_df', load_data()[0])
cache.set('last_trigger_time', load_data()[1])

@callback(Output('update-data-time', 'children'),
          Input('interval-component', 'n_intervals'))
def update_data(n):
    current_time = dt.datetime.now()
    last_trigger_time = cache.get('last_trigger_time')
    print(f"updata data time is {last_trigger_time}")
    
    if current_time.minute % 5 == 0:
        new_df, new_time = load_data()
        cache.set('data_df', new_df)
        cache.set('last_trigger_time', new_time)
        print("Callback triggered at every 5 minutes")
        print(f"Last trigger time is {new_time}")
        return f"last data update time is {new_time}"
    else:
        print(f"Waiting for the next 5 minutes...")
        print(f"Last trigger time is {last_trigger_time}")
        return f"last data update time is {last_trigger_time}"

3. 单进程模式下使用全局变量（不推荐） #

如果你的Dash应用以单进程运行（关闭debug模式或设置processes=1），可以使用global关键字临时解决，但多进程环境下每个进程会有独立的全局变量副本，导致数据不一致，官方不推荐此方法。

修改后的回调：

data = dict()
data['df'], data['last_trigger_time'] = load_data()

@callback(Output('update-data-time', 'children'),
          Input('interval-component', 'n_intervals'))
def update_data(n):
    global data
    current_time = dt.datetime.now()
    print(f"updata data time is {data['last_trigger_time']}")
    if (current_time.minute % 5 == 0):
        data['df'], data['last_trigger_time'] = load_data()
        print("Callback triggered at every 5 minutes")
    else:
        print(f"Waiting for the next 5 minutes...")
    print(f"Last trigger time is {data['last_trigger_time']}")
    return f"last data update time is {data['last_trigger_time']}"

# 启动时指定单进程
if __name__ == '__main__':
    app.run_server(debug=False, processes=1)

内容的提问来源于stack exchange，提问作者Li Hung Chun