PyShiny中如何基于用户输入的数据库路径定期更新数据？

问题场景 #

编写PyShiny应用时遇到以下问题：

• 应用需从SQL数据库读取数据（使用pandas.read_sql），数据库路径由用户通过ui.input_file上传提供
• 希望通过invalidateLater实现数据定期更新，点击Go按钮后每5秒计算数据库表table1的shape
• 运行时发现，即使修改数据库数据，显示的shape始终是首次计算的初始值

用户原始代码：

import pandas as pd
import sqlite3

from shiny import App, render, ui, reactive

app_ui = ui.page_fluid(
    ui.input_file("file1", "Choose a file to upload:", multiple=False),
    ui.input_action_button(id="go", label="Go", class_="btn-warning"),
    ui.output_text("output_txt"),
)


def server(input, output, session):
    def get_data():
        if input.go() > 0:
            reactive.invalidate_later(5)
            cnx = sqlite3.connect(input.file1()[0]["datapath"])
            df = pd.read_sql("SELECT * FROM table1", cnx)
            cnx.close()
            return df

    @output(id="output_txt")
    @render.text
    def output():
        d = get_data()
        print(d.shape)
        return d.shape


app = App(app_ui, server)

问题原因 #

1. get_data不是响应式函数：当前get_data是普通函数，未被标记为响应式依赖追踪对象，Shiny无法感知到它需要重新执行。reactive.invalidate_later()必须在响应式上下文（如@reactive.calc或@reactive.Effect装饰的函数）中使用才能生效。
2. 响应式依赖未正确触发：点击Go按钮后，虽然调用了invalidate_later，但由于get_data不是响应式计算，后续的失效信号无法触发它重新执行，导致一直返回第一次读取的结果。

解决建议与修正代码 #

将get_data转换为响应式计算函数，确保invalidate_later在响应式上下文中执行，同时补充边界情况处理避免报错：

import pandas as pd
import sqlite3

from shiny import App, render, ui, reactive

app_ui = ui.page_fluid(
    ui.input_file("file1", "Choose a file to upload:", multiple=False),
    ui.input_action_button(id="go", label="Go", class_="btn-warning"),
    ui.output_text("output_txt"),
)


def server(input, output, session):
    @reactive.calc
    def get_data():
        # 未点击Go或未上传文件时返回空值
        if input.go() == 0 or not input.file1():
            return None
        
        # 每5秒让当前计算失效，触发重新执行
        reactive.invalidate_later(5)
        
        # 连接数据库并读取最新数据
        cnx = sqlite3.connect(input.file1()[0]["datapath"])
        df = pd.read_sql("SELECT * FROM table1", cnx)
        cnx.close()
        return df

    @output(id="output_txt")
    @render.text
    def output():
        d = get_data()
        if d is None:
            return "请上传文件并点击Go按钮"
        print(d.shape)
        return f"表table1的shape: {d.shape}"


app = App(app_ui, server)

关键改动说明 #

• 用@reactive.calc装饰get_data，将其转为响应式计算函数，Shiny会自动追踪依赖，在依赖变化或定时失效时重新执行函数
• 添加边界判断，避免未上传文件或未点击Go时出现索引错误
• 确保reactive.invalidate_later(5)在响应式计算函数内部执行，每5秒触发一次重新读取数据库操作

内容的提问来源于stack exchange，提问作者Tristan Kleyn