You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python中为含API调用的方法实现可配置刷新时间的缓存(无第三方库)

给API方法实现可配置刷新时间的缓存(无第三方库)

要实现这个需求,我们可以在类内部维护缓存数据和过期时间戳,通过判断当前时间与过期时间的关系来决定是否使用缓存或重新请求API。具体步骤如下:

1. 初始化缓存相关属性

首先在类的__init__方法里添加缓存存储变量、过期时间记录,以及可配置的刷新间隔参数:

def __init__(self, config):
    self.config = config
    # 存储缓存的流量数据
    self._cached_flows = None
    # 缓存的过期时间戳(秒级)
    self._cache_expire_at = 0
    # 从配置读取刷新时间,默认5分钟(300秒)
    self.cache_refresh_secs = getattr(config, "cache_refresh_seconds", 300)

2. 修改get_flows方法加入缓存逻辑

引入time模块处理时间判断,在方法里先检查缓存是否有效,无效则重新拉取数据并更新缓存:

import time

def get_flows(self):
    now = time.time()
    # 缓存存在且未过期,直接返回缓存
    if self._cached_flows is not None and now < self._cache_expire_at:
        return self._cached_flows
    
    # 缓存失效,重新调用API获取数据
    fresh_flows = self._ans_api.get_ans(ans_name=self.config.ans).content
    # 更新缓存和过期时间
    self._cached_flows = fresh_flows
    self._cache_expire_at = now + self.cache_refresh_secs
    return fresh_flows

3. 配置刷新时间

你可以通过类的配置对象config传入自定义的刷新间隔,比如在配置里设置cache_refresh_seconds=600(10分钟);如果配置里没有这个字段,代码会默认使用300秒的刷新时间。

可选优化

  • 如果代码运行在多线程环境下,可以给缓存操作加锁,避免多个线程同时触发API请求:
import threading

def __init__(self, config):
    # 其他初始化代码...
    self._cache_lock = threading.Lock()

def get_flows(self):
    now = time.time()
    with self._cache_lock:
        if self._cached_flows is not None and now < self._cache_expire_at:
            return self._cached_flows
        
        fresh_flows = self._ans_api.get_ans(ans_name=self.config.ans).content
        self._cached_flows = fresh_flows
        self._cache_expire_at = now + self.cache_refresh_secs
        return fresh_flows
  • 还可以添加主动清空缓存的方法,方便需要时强制刷新:
def clear_flows_cache(self):
    with self._cache_lock:
        self._cached_flows = None
        self._cache_expire_at = 0

内容的提问来源于stack exchange,提问作者LearnerJS

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.20 04:01:08