如何定期刷新Flask应用的数据库数据及相关全局变量？

实现思路 #

不需要重启Flask服务，通过后台定时任务定期执行数据拉取、计算逻辑，直接更新全局变量即可。用到APScheduler库实现后台定时调度，不会阻塞Flask的请求处理循环。

前置依赖 #

首先安装需要的第三方包：

pip install flask pandas apscheduler

完整可运行示例代码（5分钟自动刷新） #

from flask import Flask, render_template
import pandas as pd
import datetime as dt
from apscheduler.schedulers.background import BackgroundScheduler

# 定义全局变量存储预计算的数据
df = None
var1 = None
var2 = None

def refresh_data():
    """数据刷新逻辑，原代码中需要刷新的部分都放在这个函数里"""
    global df, var1, var2
    # 实际使用时这里替换成从Oracle数据库拉取数据的SQL逻辑
    df = pd.DataFrame(range(5), columns=['Column1'])
    df['Column2'] = df['Column1'] * 2
    df['Column3'] = dt.datetime.now()
    df['Column4'] = df['Column3'] + pd.to_timedelta(df['Column2'], 'd')
    var1 = min(df['Column4'])
    var2 = max(df['Column4'])
    print(f"数据已刷新，当前时间：{dt.datetime.now()}")

# 应用启动时先执行一次数据加载
refresh_data()

# 初始化Flask应用
app = Flask(__name__, static_url_path='',
            static_folder='static',
            template_folder='template')
app.config["DEBUG"] = True

@app.route("/home")
def home():
    return render_template("home.html",
                           var1=str(var1),
                           var2=str(var2)) 

if __name__ == '__main__':
    # 初始化后台定时调度器
    scheduler = BackgroundScheduler()
    # 添加定时任务：每5分钟执行一次refresh_data，可改为86400秒对应每天一次
    scheduler.add_job(refresh_data, 'interval', minutes=5)
    scheduler.start()
    try:
        app.run(host="localhost", port=5000, debug=True)
    except (KeyboardInterrupt, SystemExit):
        # 关闭Flask时同时停止调度器
        scheduler.shutdown()

配套home.html模板 #

在项目的template文件夹下新建home.html，内容如下：

<HTML>
    <body>
        <div>This is Var1: {{var1}} </div>
        <div>This is Var2: {{var2}} </div>
    </body>
</HTML>

注意事项 #

• 单进程调试/单worker部署场景下直接可用，若使用多worker部署（如uWSGI、Gunicorn开多进程），需要额外配置保证调度器只在一个进程中启动，避免重复执行数据刷新任务
• 若数据计算耗时较长，可在refresh_data函数中加线程锁，避免请求读取到更新一半的脏数据

内容的提问来源于stack exchange，提问作者bart cubrich