Python不可靠网络下：lru_cache缓存失效时返回过期数据的可行方案

原生functools.lru_cache确实没有内置的失败回退机制，它只负责缓存函数返回值，对函数执行成功或失败完全无感知。要实现“超时后获取新值失败则返回过期数据”的需求，得在它外面做一层封装，或者自己实现带容错逻辑的缓存。

方案1：自定义带失败回退的缓存装饰器 #

如果不需要LRU淘汰机制，直接自己维护缓存，同时处理超时和失败场景：

import time
from functools import wraps

def cache_with_stale_fallback(max_age=300):
    """缓存超时后尝试获取新值，失败则返回旧值的装饰器"""
    _cache = {}  # 存储格式：key -> (value, timestamp)

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成唯一缓存键，兼容位置参数和关键字参数
            cache_key = (args, frozenset(kwargs.items()))
            current_time = time.time()

            if cache_key in _cache:
                cached_val, cached_time = _cache[cache_key]
                # 缓存未超时，直接返回
                if current_time - cached_time < max_age:
                    return cached_val
                # 缓存超时，尝试获取新值
                try:
                    new_val = func(*args, **kwargs)
                    _cache[cache_key] = (new_val, current_time)
                    return new_val
                except Exception as e:
                    # 获取新值失败，返回旧缓存
                    # 建议只捕获后端API相关异常，避免掩盖代码自身bug
                    return cached_val
            # 首次调用无缓存，直接请求并缓存
            try:
                val = func(*args, **kwargs)
                _cache[cache_key] = (val, current_time)
                return val
            except Exception:
                # 首次调用失败无旧值可返回，只能抛出异常
                raise
        return wrapper
    return decorator

使用示例： #

@cache_with_stale_fallback(max_age=600)  # 设置10分钟缓存超时
def fetch_backend_data(webhook_payload):
    # 此处编写调用后端API的逻辑，请求失败时抛出异常
    response = requests.post("https://your-backend-api.com/webhook", json=webhook_payload)
    response.raise_for_status()
    return response.json()

方案2：保留LRU淘汰的封装方案 #

如果需要LRU内存淘汰机制（避免缓存无限占用内存），可以在lru_cache外层套失败回退逻辑，同时单独保存一份“保底”旧值：

import time
from functools import lru_cache, wraps

def stale_lru_cache(max_age=300, lru_maxsize=128):
    """结合LRU缓存和失败回退的装饰器"""
    _stale_cache = {}  # 保存最近成功的结果，用于失败时回退

    def decorator(func):
        # 先用lru_cache处理LRU淘汰，若需TTL超时可替换为cachetools.TTLCache
        @lru_cache(maxsize=lru_maxsize)
        def _cached_func(*args, **kwargs):
            result = func(*args, **kwargs)
            cache_key = (args, frozenset(kwargs.items()))
            _stale_cache[cache_key] = result
            return result

        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return _cached_func(*args, **kwargs)
            except Exception:
                cache_key = (args, frozenset(kwargs.items()))
                # 调用失败时返回最近成功的旧值
                if cache_key in _stale_cache:
                    return _stale_cache[cache_key]
                # 无旧值可返回，抛出原异常
                raise
        return wrapper
    return decorator

补充提示 #

• 若使用带TTL的LRU缓存库（如cachetools.TTLCache），直接替换上述代码中的lru_cache即可，逻辑保持一致
• 异常捕获要精准，比如只捕获requests.exceptions.RequestException这类后端API请求异常，避免掩盖代码自身的逻辑错误
• 若需限制旧值的保留时长，可以在_stale_cache中同时存储时间戳，定期清理超期的旧值

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