如何在Python Dash回调间持久化数据实现告警功能

我在首页设置了若干徽章，用颜色标识后台进程状态，通过dcc.Interval每30秒从Redis拉取进程信息更新徽章颜色。现在需要实现徽章变为红色（danger）时触发告警音，但难点是如何保存上一次的颜色数据，以便和新生成的颜色列表对比，且避免重复查询Redis。

当前回调代码仅能更新徽章颜色，无法留存历史状态：

def initialise_callbacks(app):
    @app.callback(
        [Output("{}".format(file), "color") for file in files],
        [Input("live-update2", "n_intervals")],
    )
    def update_greeks(interval):

        # default to list of "danger"
        color_list = ["danger" for i in files]

        i = 0
        for file in files:
            update_time = conn.get("{}_update".format(file.upper())) #get status from database
            if update_time:
                if update_time.isOld: 
                    color_list[i] = "success"
                else:
                    color_list[i] = "danger"
            else:
                color_list[i] = "danger"
        i = i + 1
        return color_list

尝试过用dcc.Store但遇到同一变量既做输入又做输出的报错，不想用两个回调重复查询Redis。

可以通过单个回调同时处理Redis查询、状态对比、徽章更新和存储更新，利用dcc.Store的State获取旧状态，同时将其作为Output更新为新状态，避免重复查询。

步骤1：添加dcc.Store组件 #

在布局中添加一个dcc.Store用于存储历史颜色列表，设置初始值为全"danger"的列表：

dcc.Store(id='color-history-store', data=["danger"]*len(files))

步骤2：修改回调逻辑 #

更新回调，同时接收dcc.Store的旧状态作为State，并将新状态作为Output之一。在回调内完成一次Redis查询，生成新颜色列表后和旧列表对比，触发告警逻辑，最后返回徽章颜色和新的存储数据：

def initialise_callbacks(app):
    @app.callback(
        # 输出：所有徽章的颜色 + 存储的新状态
        [Output(f"{file}", "color") for file in files] + [Output('color-history-store', 'data')],
        # 输入：定时器触发
        [Input("live-update2", "n_intervals")],
        # 状态：上一次的颜色列表
        [State('color-history-store', 'data')]
    )
    def update_greeks(interval, old_color_list):
        # 1. 单次Redis查询生成新颜色列表
        new_color_list = ["danger" for _ in files]
        for idx, file in enumerate(files):
            update_time = conn.get(f"{file.upper()}_update")
            if update_time:
                new_color_list[idx] = "success" if update_time.isOld else "danger"
        
        # 2. 对比新旧状态，触发告警逻辑
        # 遍历检查每个徽章是否从非danger变为danger
        for idx, (old_color, new_color) in enumerate(zip(old_color_list, new_color_list)):
            if old_color != "danger" and new_color == "danger":
                # 此处添加告警音触发逻辑，例如调用前端音频播放函数
                play_alert_sound()
        
        # 3. 返回徽章颜色和新的存储数据
        return new_color_list + [new_color_list]

关键说明 #

• 仅需一次Redis查询即可获取所有进程状态，避免重复操作；
• dcc.Store同时作为State（获取旧值）和Output（更新新值）在Dash中是合法的，只要不同时将其设为Input和Output即可；
• 告警逻辑可根据需求实现，比如通过clientside_callback在前端直接播放音频，减少后端交互。

内容的提问来源于stack exchange，提问作者Marcelo ARK