pip install cachetools

from cachetools import LRUCache, TTLCache, LFUCache

# 創(chuàng)建 LRU 緩存，最大容量為 5
lru_cache = LRUCache(maxsize=5)

# 創(chuàng)建 TTL 緩存，最大容量為 10，每項(xiàng)存活時(shí)間為 60 秒
ttl_cache = TTLCache(maxsize=10, ttl=60)

# 創(chuàng)建 LFU 緩存，最大容量為 3
lfu_cache = LFUCache(maxsize=3)

from cachetools import cached, TTLCache

# 創(chuàng)建一個(gè) TTL 緩存
cache = TTLCache(maxsize=3, ttl=10)

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

# 第一次調(diào)用會(huì)計(jì)算并緩存
print(expensive_computation(4))  # 輸出: 16
# 第二次調(diào)用直接從緩存獲取
print(expensive_computation(4))  # 輸出: 16（無(wú)計(jì)算）

from cachetools import LRUCache

cache = LRUCache(maxsize=2)

# 添加數(shù)據(jù)到緩存中
cache['a'] = 1
cache['b'] = 2

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

# 添加新數(shù)據(jù)，超出容量時(shí)移除最少使用的數(shù)據(jù)
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))  # 實(shí)際調(diào)用
print(fetch_data(1, 2, z=3))  # 從緩存中獲取

import requests
from cachetools import cached, TTLCache

# 創(chuàng)建一個(gè) TTL 緩存，最大容量為 5，每項(xiàng)生存時(shí)間為 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))   # 實(shí)際請(qǐng)求 API
print(fetch_api_data(url))   # 從緩存獲取結(jié)果

from cachetools import TTLCache
import time
 
class CallbackTTLCache(TTLCache):
    """支持過(guò)期回調(diào)的 TTL 緩存"""
    def __init__(self, maxsize, ttl, callback=None):
        super().__init__(maxsize, ttl)
        self.callback = callback  # 過(guò)期回調(diào)函數(shù)
        self._pending_callbacks = {}  # 跟蹤需要回調(diào)的項(xiàng)目
 
    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):
        """淘汰條目時(shí)觸發(fā)回調(diào)"""
        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("緩存已經(jīng)為空")
            return None, None
 
    def clear_expired(self):
        """清理所有過(guò)期的項(xiàng)目并觸發(fā)回調(diào)"""
        current_time = time.time()
        expired_keys = []
        
        # 檢查所有鍵是否過(guò)期
        for key in list(self._pending_callbacks):
            try:
                # 嘗試獲取值，如果過(guò)期會(huì)拋出KeyError
                _ = self[key]
            except KeyError:
                # 如果拋出KeyError，說(shuō)明已過(guò)期
                expired_keys.append(key)
        
        # 觸發(fā)過(guò)期回調(diào)
        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):
        """清空緩存并觸發(fā)所有回調(diào)"""
        for key, value in list(self._pending_callbacks.items()):
            if self.callback:
                self.callback(key, value)
        self._pending_callbacks.clear()
        super().clear()
 
# 示例：定義過(guò)期回調(diào)函數(shù)
def on_expired(key, value):
    print(f"?? 緩存過(guò)期！Key: {key}, Value: {value} 已清除")
 
# 初始化緩存（最大100條，有效期5秒，綁定回調(diào)）
cache = CallbackTTLCache(maxsize=100, ttl=5, callback=on_expired)
 
# 測(cè)試用例
def run_tests():
    print("=== 開(kāi)始測(cè)試 ===")
    
    # 測(cè)試1：添加并立即訪問(wèn)緩存
    print("\n測(cè)試1：添加并立即訪問(wèn)緩存")
    cache["user_101"] = "Alice"
    cache["user_102"] = "Bob"
    print("初始緩存內(nèi)容:", dict(cache.items()))
    
    # 測(cè)試2：等待部分時(shí)間后訪問(wèn)
    print("\n測(cè)試2：等待3秒后訪問(wèn)緩存")
    time.sleep(3)
    print("訪問(wèn)user_101:", cache.get("user_101", "已過(guò)期"))
    print("3秒后緩存內(nèi)容:", dict(cache.items()))
    
    # 測(cè)試3：等待過(guò)期并嘗試訪問(wèn)
    print("\n測(cè)試3：等待剩余時(shí)間直到過(guò)期")
    time.sleep(3)  # 再等3秒，總共6秒，確保過(guò)期
    print("嘗試訪問(wèn)user_101:", cache.get("user_101", "已過(guò)期"))
    print("6秒后（過(guò)期）緩存內(nèi)容:", dict(cache.items()))
    
    # 測(cè)試4：手動(dòng)清理過(guò)期內(nèi)容
    print("\n測(cè)試4：清理過(guò)期內(nèi)容")
    cache.clear_expired()
    print("清理后緩存內(nèi)容:", dict(cache.items()))
    
    # 測(cè)試5：嘗試對(duì)空緩存調(diào)用popitem
    print("\n測(cè)試5：對(duì)空緩存調(diào)用popitem")
    key, value = cache.popitem()
    print(f"popitem結(jié)果 - Key: {key}, Value: {value}")
    
    print("\n=== 測(cè)試完成 ===")
 
if __name__ == "__main__":
    run_tests()

import time
from functools import wraps
 
def timed_cache(ttl=60, callback=None):
    """支持過(guò)期時(shí)間和回調(diào)的緩存裝飾器"""
    cache = {}
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, frozenset(kwargs.items()))
            current_time = time.time()
            
            # 檢查緩存是否命中且未過(guò)期
            if key in cache:
                result, expire_time = cache[key]
                if current_time < expire_time:
                    return result
                elif callback:  # 觸發(fā)過(guò)期回調(diào)
                    callback(key, result)
            
            # 重新計(jì)算并緩存結(jié)果
            result = func(*args, **kwargs)
            cache[key] = (result, current_time + ttl)
            return result
        return wrapper
    return decorator
 
# 定義回調(diào)函數(shù)
def expire_callback(key, value):
    print(f"?? 回調(diào)通知: 參數(shù) {key} 的值 {value} 已過(guò)期")
 
# 應(yīng)用裝飾器
@timed_cache(ttl=3, callback=expire_callback)
def heavy_compute(x):
    print(f"計(jì)算中... 參數(shù)={x}")
    return x ** 2
 
# 測(cè)試
print(heavy_compute(4))  # 首次計(jì)算，輸出: 計(jì)算中... 16
print(heavy_compute(4))  # 命中緩存，無(wú)輸出 → 16
time.sleep(4)
print(heavy_compute(4))  # 過(guò)期后重新計(jì)算，觸發(fā)回調(diào) → 計(jì)算中... 16