Flask Cache时间范围复用问题：非时段无法读取历史缓存数据

阿华AIGC实验室

2026-5-20

解决Flask应用特定时段拉取数据+非时段复用缓存的问题

看起来你遇到的核心问题是缓存依赖用户访问维持，而非主动提前缓存，导致非时段如果很久没人访问接口就没数据可用。下面我给你一套完整的解决方案，结合定时任务主动预缓存+时段判断逻辑，完美解决你的需求：

核心思路

用定时任务在目标URL停止返回数据前1分钟（也就是每天16:29）主动拉取并缓存最新数据，确保非时段有数据可用
在接口中加入时段判断：
- 允许时段内：优先实时拉取数据，失败则 fallback 到缓存
- 非允许时段：直接返回预缓存的数据
缓存配置成长期有效（直到下一次定时任务更新），避免自动过期导致数据丢失

实现步骤

1. 安装依赖

首先安装需要的包：

pip install flask flask-caching apscheduler

2. 完整代码示例

from flask import Flask, jsonify
from flask_caching import Cache
from apscheduler.schedulers.background import BackgroundScheduler
import requests
from datetime import datetime

app = Flask(__name__)

# 配置缓存：用文件系统缓存，也可以换成Redis
app.config['CACHE_TYPE'] = 'FileSystemCache'
app.config['CACHE_DIR'] = './cache'
app.config['CACHE_DEFAULT_TIMEOUT'] = 86400  # 默认缓存1天，足够覆盖非时段

cache = Cache(app)

# 目标数据URL和时段配置
TARGET_URL = "你的指定数据URL"
# 允许实时拉取的时段：比如每天9:00到16:30
ALLOWED_START_HOUR = 9
ALLOWED_END_HOUR = 16
ALLOWED_END_MINUTE = 30

def fetch_and_cache_data():
    """拉取最新数据并缓存，供定时任务调用"""
    try:
        response = requests.get(TARGET_URL)
        response.raise_for_status()  # 检查请求是否成功
        data = response.json()
        cache.set('top_data', data)
        print(f"[{datetime.now()}] 成功缓存最新数据")
    except Exception as e:
        print(f"[{datetime.now()}] 缓存数据失败：{str(e)}")
        # 如果拉取失败，可以考虑保留旧缓存，不做操作

def is_in_allowed_time():
    """判断当前是否在允许实时拉取数据的时段内"""
    now = datetime.now()
    # 先判断小时：如果在9-15点，直接允许；如果是16点，需要分钟小于30
    if now.hour < ALLOWED_START_HOUR or now.hour > ALLOWED_END_HOUR:
        return False
    if now.hour == ALLOWED_END_HOUR and now.minute >= ALLOWED_END_MINUTE:
        return False
    return True

@app.route('/top')
def get_top_data():
    # 先尝试获取缓存数据
    cached_data = cache.get('top_data')
    
    # 判断当前是否在允许时段内
    if is_in_allowed_time():
        try:
            # 实时拉取最新数据
            response = requests.get(TARGET_URL)
            response.raise_for_status()
            real_time_data = response.json()
            # 更新缓存
            cache.set('top_data', real_time_data)
            return jsonify(real_time_data)
        except Exception as e:
            # 实时拉取失败，返回缓存数据（如果有的话）
            if cached_data:
                print(f"实时拉取失败，返回缓存数据：{str(e)}")
                return jsonify(cached_data)
            else:
                return jsonify({"error": "实时拉取失败且无缓存数据"}), 500
    else:
        # 非允许时段，直接返回缓存数据
        if cached_data:
            return jsonify(cached_data)
        else:
            return jsonify({"error": "无缓存数据，请在允许时段访问后重试"}), 500

# 初始化定时任务：每天16:29执行缓存操作
scheduler = BackgroundScheduler()
# 添加定时任务，每天16:29运行fetch_and_cache_data
scheduler.add_job(fetch_and_cache_data, 'cron', hour=16, minute=29)
scheduler.start()

if __name__ == '__main__':
    # 启动前先手动缓存一次（可选，避免第一天启动时无缓存）
    fetch_and_cache_data()
    app.run(debug=True)