Python cachetools实现缓存过期策略

pip install cachetools

from cachetools import LRUCache, TTLCache, LFUCache

# 创建 LRU 缓存，最大容量为 5
lru_cache = LRUCache(maxsize=5)

# 创建 TTL 缓存，最大容量为 10，每项存活时间为 60 秒
ttl_cache = TTLCache(maxsize=10, ttl=60)

# 创建 LFU 缓存，最大容量为 3
lfu_cache = LFUCache(maxsize=3)

from cachetools import cached, TTLCache

# 创建一个 TTL 缓存
cache = TTLCache(maxsize=3, ttl=10)

@cached(cache)
def expensive_computation(x):
    print(f"Computing {x}...")
    return x * x

# 第一次调用会计算并缓存
print(expensive_computation(4))  # 输出: 16
# 第二次调用直接从缓存获取
print(expensive_computation(4))  # 输出: 16（无计算）

from cachetools import LRUCache

cache = LRUCache(maxsize=2)

# 添加数据到缓存中
cache['a'] = 1
cache['b'] = 2

print(cache)  # 输出: {'a': 1, 'b': 2}

# 添加新数据，超出容量时移除最少使用的数据
cache['c'] = 3
print(cache)  # 输出: {'b': 2, 'c': 3}

from cachetools import cached, LRUCache, keys

def custom_key(*args, **kwargs):
    return keys.hashkey(*args) + tuple(sorted(kwargs.items()))

cache = LRUCache(maxsize=128)

@cached(cache, key=custom_key)
def fetch_data(x, y, z=None):
    print(f"Fetching data for {x}, {y}, {z}")
    return x + y + (z or 0)

print(fetch_data(1, 2, z=3))  # 实际调用
print(fetch_data(1, 2, z=3))  # 从缓存中获取

import requests
from cachetools import cached, TTLCache

# 创建一个 TTL 缓存，最大容量为 5，每项生存时间为 30 秒
api_cache = TTLCache(maxsize=5, ttl=30)

@cached(api_cache)
def fetch_api_data(url):
    print(f"Fetching data from {url}")
    response = requests.get(url)
    return response.json() if response.status_code == 200 else None

url = "https://jsonplaceholder.typicode.com/todos/1"
print(fetch_api_data(url))   # 实际请求 API
print(fetch_api_data(url))   # 从缓存获取结果

from cachetools import TTLCache
import time
 
class CallbackTTLCache(TTLCache):
    """支持过期回调的 TTL 缓存"""
    def __init__(self, maxsize, ttl, callback=None):
        super().__init__(maxsize, ttl)
        self.callback = callback  # 过期回调函数
        self._pending_callbacks = {}  # 跟踪需要回调的项目
 
    def __setitem__(self, key, value, **kwargs):
        super().__setitem__(key, value, **kwargs)
        if self.callback:
            self._pending_callbacks[key] = value
 
    def __getitem__(self, key):
        try:
            return super().__getitem__(key)
        except KeyError:
            if key in self._pending_callbacks:
                value = self._pending_callbacks.pop(key)
                if self.callback:
                    self.callback(key, value)
            raise
 
    def popitem(self):
        """淘汰条目时触发回调"""
        try:
            key, value = super().popitem()
            if key in self._pending_callbacks:
                self._pending_callbacks.pop(key)
            if self.callback:
                self.callback(key, value)
            return key, value
        except KeyError:
            print("缓存已经为空")
            return None, None
 
    def clear_expired(self):
        """清理所有过期的项目并触发回调"""
        current_time = time.time()
        expired_keys = []
        
        # 检查所有键是否过期
        for key in list(self._pending_callbacks):
            try:
                # 尝试获取值，如果过期会抛出KeyError
                _ = self[key]
            except KeyError:
                # 如果抛出KeyError，说明已过期
                expired_keys.append(key)
        
        # 触发过期回调
        for key in expired_keys:
            if key in self._pending_callbacks:
                value = self._pending_callbacks.pop(key)
                if self.callback:
                    self.callback(key, value)
 
    def clear(self):
        """清空缓存并触发所有回调"""
        for key, value in list(self._pending_callbacks.items()):
            if self.callback:
                self.callback(key, value)
        self._pending_callbacks.clear()
        super().clear()
 
# 示例：定义过期回调函数
def on_expired(key, value):
    print(f"🚨 缓存过期！Key: {key}, Value: {value} 已清除")
 
# 初始化缓存（最大100条，有效期5秒，绑定回调）
cache = CallbackTTLCache(maxsize=100, ttl=5, callback=on_expired)
 
# 测试用例
def run_tests():
    print("=== 开始测试 ===")
    
    # 测试1：添加并立即访问缓存
    print("\n测试1：添加并立即访问缓存")
    cache["user_101"] = "Alice"
    cache["user_102"] = "Bob"
    print("初始缓存内容:", dict(cache.items()))
    
    # 测试2：等待部分时间后访问
    print("\n测试2：等待3秒后访问缓存")
    time.sleep(3)
    print("访问user_101:", cache.get("user_101", "已过期"))
    print("3秒后缓存内容:", dict(cache.items()))
    
    # 测试3：等待过期并尝试访问
    print("\n测试3：等待剩余时间直到过期")
    time.sleep(3)  # 再等3秒，总共6秒，确保过期
    print("尝试访问user_101:", cache.get("user_101", "已过期"))
    print("6秒后（过期）缓存内容:", dict(cache.items()))
    
    # 测试4：手动清理过期内容
    print("\n测试4：清理过期内容")
    cache.clear_expired()
    print("清理后缓存内容:", dict(cache.items()))
    
    # 测试5：尝试对空缓存调用popitem
    print("\n测试5：对空缓存调用popitem")
    key, value = cache.popitem()
    print(f"popitem结果 - Key: {key}, Value: {value}")
    
    print("\n=== 测试完成 ===")
 
if __name__ == "__main__":
    run_tests()

import time
from functools import wraps
 
def timed_cache(ttl=60, callback=None):
    """支持过期时间和回调的缓存装饰器"""
    cache = {}
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, frozenset(kwargs.items()))
            current_time = time.time()
            
            # 检查缓存是否命中且未过期
            if key in cache:
                result, expire_time = cache[key]
                if current_time < expire_time:
                    return result
                elif callback:  # 触发过期回调
                    callback(key, result)
            
            # 重新计算并缓存结果
            result = func(*args, **kwargs)
            cache[key] = (result, current_time + ttl)
            return result
        return wrapper
    return decorator
 
# 定义回调函数
def expire_callback(key, value):
    print(f"🔔 回调通知: 参数 {key} 的值 {value} 已过期")
 
# 应用装饰器
@timed_cache(ttl=3, callback=expire_callback)
def heavy_compute(x):
    print(f"计算中... 参数={x}")
    return x ** 2
 
# 测试
print(heavy_compute(4))  # 首次计算，输出: 计算中... 16
print(heavy_compute(4))  # 命中缓存，无输出 → 16
time.sleep(4)
print(heavy_compute(4))  # 过期后重新计算，触发回调 → 计算中... 16