【python 第三方库】cachetool

2024-01-25 00:00:00

目录:

cachetools 介绍

cachetools 是一个功能强大的 Python 第三方缓存库。

$ pip install cachetools

缓存策略如下:

  • LRU(Least Recently Used,最近最少使用):当缓存达到最大容量时,会优先移除最久未被访问的条目。这是最常用且适用于大多数场景的策略。
  • TTL(Time-To-Live,生存时间):为每个缓存项设置一个过期时间,到期后自动移除。非常适合缓存具有时效性的数据,如API响应、会话信息等。
  • LFU(Least Frequently Used,最不经常使用):当缓存满时,移除访问频率最低的条目。适用于那些访问热度差异明显的场景。
  • FIFO(First-In-First-Out,先进先出):严格按照条目添加到缓存的顺序进行移除,先添加的先被淘汰。
  • RR(Random Replacement,随机替换):随机选择一个条目进行移除。实现简单,但性能相对不如其他策略。

创建不同类型的缓存

指定最大容量 (maxsize)。这个参数代表缓存中可以存储的最大条目数量,而非内存大小。

from cachetools import LRUCache, TTLCache, LFUCache, FIFOCache, RRCache

# 创建LRU缓存,最多存储 100 个条目
lru_cache = LRUCache(maxsize=100)

# 创建TTL缓存,最多存储 100 个条目,每个条目存活 60 秒
ttl_cache = TTLCache(maxsize=100, ttl=60)

# 创建LFU缓存
lfu_cache = LFUCache(maxsize=100)

# 创建FIFO缓存
fifo_cache = FIFOCache(maxsize=100)

# 创建RR缓存
rr_cache = RRCache(maxsize=100)

缓存增删改查

# 添加缓存项
lru_cache["user_id_123"] = {"name": "Alice", "age": 30}

# 获取缓存项,如果不存在则返回默认值
user_info = lru_cache.get("user_id_123", None)

# 删除缓存项
if "user_id_123" in lru_cache:
    del lru_cache["user_id_123"]

# 更新缓存项
lru_cache["user_id_123"] = {"name": "Alice", "age": 31}

TTL 缓存实现数据自动过期

import time
from cachetools import TTLCache

# 创建TTL缓存,条目存活5秒
cache = TTLCache(maxsize=3, ttl=5)
cache['a'] = 1
time.sleep(1)
cache['b'] = 2
print(cache.items())  # 输出: [('a', 1), ('b', 2)]

time.sleep(4.5)  # 总共过去5.5秒,'a'已过期
print(cache.items())  # 输出: [('b', 2)]

time.sleep(1)  # 'b'也过期
print(cache.items())  # 输出: []

函数缓存装饰器

非常实用的场景:缓存 API 请求结果,默认情况下,函数的参数会被用作缓存的键,如下函数参数 user_id 是个字符串。

import requests
from cachetools import cached, TTLCache

# 创建一个最大容量为5,TTL为30秒的缓存
api_cache = TTLCache(maxsize=5, ttl=30)

@cached(api_cache)
def fetch_user_data(user_id):
    print(f"Fetching data for user {user_id} from API...")
    response = requests.get(f" https://api.example.com/users/ {user_id}")
    return response.json() if response.status_code == 200 else None

# 第一次调用会真正请求API并缓存结果
data1 = fetch_user_data(1)
# 第二次调用(在30秒内)会直接返回缓存结果,无需网络请求
data2 = fetch_user_data(1)

如果遇到函数参数包含不可哈希的类型(如字典),需要先进行处理。

from cachetools import cached, LRUCache
import json

cache = LRUCache(maxsize=128)

def make_cache_key(api_url, params):
    # 将参数字典排序后转为 JSON 字符串,确保相同内容生成相同的键
    return api_url + json.dumps(params, sort_keys=True)

@cached(cache, key=make_cache_key)
def get_data_from_api(api_url, params):
    print(f"Calling API: {api_url} with {params}")
    response = requests.get(api_url, params=params)
    return response.json()

返回首页

本文总阅读量  次
皖ICP备17026209号-3
总访问量: 
总访客量: