キャッシュライブラリ

cachetools

概要

cachetoolsは、Pythonの拡張可能なメモ化コレクションとデコレータを提供するライブラリです。

詳細

cachetools（キャッシュツールズ）は、Python標準ライブラリのfunctools.lru_cacheを拡張し、より多様なキャッシュ戦略を提供するライブラリです。2015年にThomas Kemmerによって開発され、メモ化（関数の結果をキャッシュする技術）のためのコレクションとデコレータのセットを提供します。LRU（Least Recently Used）、FIFO（First In First Out）、TTL（Time To Live）、LFU（Least Frequently Used）、TLRUなど、多種多様なキャッシュアルゴリズムを実装しており、用途に応じて最適なキャッシュ戦略を選択できます。@cachedデコレータを使用することで、既存の関数に簡単にキャッシュ機能を追加でき、@cachedmethodデコレータではクラスメソッドのキャッシュも可能です。すべてのキャッシュクラスはcollections.MutableMappingから派生しており、辞書のような使い方ができ、maxsizeとcurrsizeプロパティでキャッシュサイズを管理できます。

メリット・デメリット

メリット

• 多様なキャッシュアルゴリズム: LRU、FIFO、TTL、LFU、TLRUなど豊富な選択肢
• デコレータ対応: @cachedと@cachedmethodで簡単にキャッシュ機能を追加
• 標準ライブラリ互換: functools.lru_cacheとの互換性
• カスタムキー関数: hashkeyやmethodkeyでカスタムキー生成が可能
• スレッドセーフティ: ロックオブジェクトと組み合わせて安全な並行処理
• 統計情報: cache_info()でヒット率やミス数を確認可能
• メモリ効率: 自動的な期限切れとサイズ制限によるメモリ管理

デメリット

• スレッドセーフティ: デフォルトではスレッドセーフではない
• メモリ使用量: 大量のデータをキャッシュする場合のメモリ消費
• 設定の複雑さ: 複数のキャッシュ戦略から最適なものを選択する必要
• TTL精度: TTLキャッシュは厳密な時間制御が難しい場合がある
• 依存関係: 外部ライブラリとしての導入コスト

主要リンク

• cachetools公式ドキュメント
• PyPI - cachetools
• GitHub - tkem/cachetools
• cachetools API リファレンス

書き方の例

基本的なキャッシュ

from cachetools import cached, LRUCache, TTLCache

# 辞書を使った基本的なメモ化
@cached(cache={})
def fibonacci(n):
    """フィボナッチ数列の計算をキャッシュ"""
    return n if n < 2 else fibonacci(n - 1) + fibonacci(n - 2)

# 使用例
print(fibonacci(100))  # 初回は計算時間がかかる
print(fibonacci(100))  # キャッシュから高速で取得

LRUキャッシュ

import urllib.request
from cachetools import cached, LRUCache

# LRUキャッシュでAPIレスポンスをキャッシュ
@cached(cache=LRUCache(maxsize=32))
def get_pep_document(num):
    """Python Enhancement Proposalの取得"""
    url = f'http://www.python.org/dev/peps/pep-{num:04d}/'
    with urllib.request.urlopen(url) as response:
        return response.read()

# 使用例
pep1 = get_pep_document(1)
pep8 = get_pep_document(8)

TTLキャッシュ（期限付き）

from cachetools import cached, TTLCache
import requests

# 10分間キャッシュするAPI呼び出し
@cached(cache=TTLCache(maxsize=1024, ttl=600))
def get_weather_data(city):
    """天気データを期限付きでキャッシュ"""
    url = f'http://api.openweathermap.org/data/2.5/weather?q={city}'
    response = requests.get(url)
    return response.json()

# 使用例
weather = get_weather_data("Tokyo")

キャッシュ統計の確認

from cachetools import cached, LRUCache

@cached(cache=LRUCache(maxsize=32), info=True)
def expensive_function(x):
    """重い計算をシミュレート"""
    import time
    time.sleep(0.1)
    return x * x

# 統計情報を確認
for i in [1, 2, 3, 1, 2, 1]:
    result = expensive_function(i)

print(expensive_function.cache_info())
# CacheInfo(hits=3, misses=3, maxsize=32, currsize=3)

クラスメソッドのキャッシュ

from cachetools import cachedmethod, LRUCache
from cachetools.keys import hashkey
import requests

class APIClient:
    def __init__(self, cache_size=100):
        self.cache = LRUCache(maxsize=cache_size)
    
    @cachedmethod(lambda self: self.cache)
    def fetch_user_data(self, user_id):
        """ユーザーデータをキャッシュ付きで取得"""
        response = requests.get(f'/api/users/{user_id}')
        return response.json()
    
    @cachedmethod(lambda self: self.cache)
    def fetch_post_data(self, post_id):
        """投稿データをキャッシュ付きで取得"""
        response = requests.get(f'/api/posts/{post_id}')
        return response.json()

# 使用例
client = APIClient()
user = client.fetch_user_data(123)
posts = client.fetch_post_data(456)

カスタムキー関数

from cachetools import cached, LRUCache
from cachetools.keys import hashkey

def custom_key(*args, config={}, **kwargs):
    """辞書を含むカスタムキー生成"""
    key = hashkey(*args, **kwargs)
    # 辞書を並べ替えてタプルに変換
    key += tuple(sorted(config.items()))
    return key

@cached(LRUCache(maxsize=128), key=custom_key)
def process_data(data, config={}):
    """設定付きのデータ処理"""
    # 設定に応じたデータ処理
    return f"Processed {data} with {config}"

# 使用例
result1 = process_data("data1", config={"format": "json", "sort": True})
result2 = process_data("data1", config={"sort": True, "format": "json"})  # 同じキーとして認識

複数関数でのキャッシュ共有

from cachetools import cached
from cachetools.keys import hashkey
from functools import partial

# 共有キャッシュ
shared_cache = {}

@cached(shared_cache, key=partial(hashkey, 'fibonacci'))
def fibonacci(n):
    return n if n < 2 else fibonacci(n - 1) + fibonacci(n - 2)

@cached(shared_cache, key=partial(hashkey, 'lucas'))
def lucas_numbers(n):
    return 2 - n if n < 2 else lucas_numbers(n - 1) + lucas_numbers(n - 2)

# 両方の関数が同じキャッシュを使用
fib_result = fibonacci(10)
lucas_result = lucas_numbers(10)
print(f"共有キャッシュサイズ: {len(shared_cache)}")