如何在Flask中持续从API获取实时数据？

下面是几个不用重启代码就能自动更新数据的实用方法，适配你的入门场景：

方案1：前端定时请求刷新（最易实现） #

直接在前端用JavaScript定时调用Flask接口，获取最新数据后更新表格，后端改动极小：

后端Flask路由（拆分数据逻辑为独立接口） #

from flask import Flask, jsonify, render_template
import requests

app = Flask(__name__)

# 封装ThingSpeak数据获取逻辑
def fetch_thingspeak_data():
    # 替换为你的ThingSpeak API地址与参数
    api_url = "https://api.thingspeak.com/channels/XXX/feeds.json?api_key=YYY&results=10"
    resp = requests.get(api_url)
    return resp.json()["feeds"] if resp.status_code == 200 else []

# 主页渲染初始表格
@app.route('/')
def index():
    return render_template('index.html', data=fetch_thingspeak_data())

# 提供实时数据的API接口
@app.route('/api/realtime-data')
def api_realtime_data():
    return jsonify(fetch_thingspeak_data())

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

前端HTML（添加定时刷新逻辑） #

在表格页面中加入这段JS，每隔5秒自动拉取并更新数据：

<table id="data-table">
    {% for item in data %}
    <tr>
        <td>{{ item.created_at }}</td>
        <td>{{ item.field1 }}</td>
        <!-- 其他字段按需添加 -->
    </tr>
    {% endfor %}
</table>

<script>
function refreshTable() {
    fetch('/api/realtime-data')
        .then(resp => resp.json())
        .then(newData => {
            const table = document.getElementById('data-table');
            // 清空原有数据行（保留表头）
            while (table.rows.length > 1) table.deleteRow(1);
            // 插入新数据
            newData.forEach(item => {
                const row = table.insertRow();
                row.insertCell().textContent = item.created_at;
                row.insertCell().textContent = item.field1;
            });
        });
}

// 页面加载后立即刷新，之后每5秒执行一次
window.onload = () => {
    refreshTable();
    setInterval(refreshTable, 5000); // 时间间隔可自定义，单位毫秒
};
</script>

方案2：后端定时缓存数据（适合需预处理数据的场景） #

如果需要在后端对数据做清洗、计算等预处理，可以用APScheduler在后台定时拉取数据并缓存，前端直接读取缓存结果：

步骤1：安装依赖 #

pip install apscheduler

步骤2：Flask集成定时任务 #

from flask import Flask, render_template
from apscheduler.schedulers.background import BackgroundScheduler
import requests

app = Flask(__name__)
# 用全局变量缓存数据（生产环境建议用Redis等专业缓存工具）
cached_data = []

def update_cached_data():
    global cached_data
    api_url = "https://api.thingspeak.com/channels/XXX/feeds.json?api_key=YYY&results=10"
    resp = requests.get(api_url)
    if resp.status_code == 200:
        cached_data = resp.json()["feeds"]

# 初始化定时任务：启动时立即执行一次，之后每5分钟更新一次
scheduler = BackgroundScheduler()
scheduler.add_job(update_cached_data, 'interval', minutes=5, id='thingspeak_sync')
scheduler.start()

@app.route('/')
def index():
    return render_template('index.html', data=cached_data)

# 关闭应用时停止定时任务
@app.teardown_appcontext
def shutdown_scheduler(exception=None):
    scheduler.shutdown()

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

方案3：Server-Sent Events（SSE，实时推送） #

如果需要数据更新时立即推送给前端，无需前端轮询，可以用SSE实现单向实时推送：

后端Flask路由 #

from flask import Flask, render_template, Response
import requests
import time

app = Flask(__name__)

def data_stream():
    while True:
        # 拉取最新一条数据
        api_url = "https://api.thingspeak.com/channels/XXX/feeds.json?api_key=YYY&results=1"
        resp = requests.get(api_url)
        if resp.status_code == 200:
            latest_item = resp.json()["feeds"][0]
            # 按SSE格式返回数据
            yield f"data: {latest_item}\n\n"
        time.sleep(3) # 每3秒检查一次更新

@app.route('/')
def index():
    return render_template('sse_index.html')

@app.route('/stream')
def stream():
    return Response(data_stream(), mimetype='text/event-stream')

前端HTML #

<table id="data-table">
    <tr>
        <th>时间</th>
        <th>字段1</th>
    </tr>
</table>

<script>
const eventSource = new EventSource('/stream');
const table = document.getElementById('data-table');

eventSource.onmessage = function(event) {
    const latestItem = JSON.parse(event.data);
    // 将新数据插入表格顶部
    const row = table.insertRow(1);
    row.insertCell().textContent = latestItem.created_at;
    row.insertCell().textContent = latestItem.field1;
};
</script>

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