ライブラリ

DiskCache

概要

DiskCacheは、Pythonでディスクベースの永続キャッシュを提供するライブラリです。メモリベースのキャッシュとは異なり、すべてのデータがディスクに保存され、プロセスや システム再起動後もデータが保持されます。

詳細

DiskCache（python-diskcache）は、SQLAlchemyプロジェクトの一部として開発された、高性能なディスクベースキャッシュライブラリです。RedisやMemcachedなどのメモリベースキャッシュの代替として設計され、特に読み取り操作で卓越したパフォーマンスを発揮します。ハイブリッドストレージアーキテクチャを採用し、キャッシュメタデータ（キー、有効期限、タグ）をSQLiteデータベースに、大きな値をファイルシステム上のファイルとして保存します。小さな値は直接SQLiteデータベース内に格納されます。FanoutCacheによる自動シャーディング機能により、高い並行性を持つ環境での書き込み性能を向上させます。DjangoCache、Deque、Index などの永続データ構造も提供し、Webアプリケーションからデータサイエンスまで幅広い用途で活用できます。

メリット・デメリット

メリット

• 永続性: プロセス再起動やシステム障害後もデータが保持される
• 高い読み取り性能: 単一プロセスでMemcachedやRedisを上回る読み取り速度
• スケーラビリティ: FanoutCacheによる並行書き込み性能の向上
• Pure Python: 外部依存なしで簡単にインストール・導入可能
• Django統合: Django標準のキャッシュバックエンドとして利用可能
• 豊富な機能: memoize装飾子、分散ロック、関数スロットリングなど
• 柔軟な設定: 複数の削除ポリシー（LRU、LFU等）をサポート

デメリット

• 書き込み性能: ディスク永続化のため、メモリベースキャッシュより書き込みが遅い
• NFSの制限: SQLiteの制約によりNFSなどのネットワークファイルシステムでは性能が低下
• ディスク使用量: 大量のデータをキャッシュする場合、ディスク容量の監視が必要
• 非同期サポート: SQLiteモジュールの制限により、ネイティブな非同期サポートがない
• 書き込み競合: シャーディングなしでは並行書き込みで高いレイテンシが発生する可能性

主要リンク

• DiskCache公式ドキュメント
• GitHub リポジトリ
• PyPI パッケージ
• ベンチマーク結果
• SQLAlchemyプロジェクト
• Django統合ガイド

書き方の例

基本的な使用方法

import diskcache as dc

# キャッシュインスタンスの作成
cache = dc.Cache('tmp')

# 値の設定と取得
cache['key'] = 'value'
value = cache['key']
print(value)  # 'value'

# キーの存在確認
if 'key' in cache:
    print('Key exists')

# キーの削除
del cache['key']

高度な設定オプション

import diskcache as dc

# 設定付きキャッシュの作成
cache = dc.Cache(
    directory='my_cache',
    size_limit=int(1e9),  # 1GBのサイズ制限
    timeout=60.0,  # デフォルトタイムアウト
    eviction_policy='least-recently-used'  # LRU削除ポリシー
)

# 有効期限付きでデータを設定
cache.set('temp_key', 'temporary_value', expire=30)  # 30秒後に期限切れ

# タグ付きでデータを設定
cache.set('tagged_key', 'data', tag='group1')

# タグによるデータ削除
cache.evict('group1')

FanoutCacheによる並行性向上

import diskcache as dc

# 4つのシャードを持つFanoutCache
cache = dc.FanoutCache(
    shards=4,
    timeout=1.0,
    directory='fanout_cache'
)

# 並行書き込みに最適化された操作
cache['key1'] = 'value1'
cache['key2'] = 'value2'

# memoize装飾子の使用
@cache.memoize(expire=300, typed=True)
def expensive_function(x, y):
    # 重い計算処理
    import time
    time.sleep(1)  # 処理時間をシミュレート
    return x * y + x ** y

# 初回は計算が実行される
result1 = expensive_function(2, 3)
# 2回目以降はキャッシュから取得される
result2 = expensive_function(2, 3)

Django統合

# settings.py
CACHES = {
    'default': {
        'BACKEND': 'diskcache.DjangoCache',
        'LOCATION': '/var/tmp/django_cache',
        'TIMEOUT': 300,
        'OPTIONS': {
            'size_limit': 2 ** 30,  # 1GB
            'cull_limit': 10,
        },
    }
}

# Djangoビューでの使用
from django.core.cache import cache
from django.views.decorators.cache import cache_page

@cache_page(60 * 15)  # 15分間キャッシュ
def my_view(request):
    # 重いデータベース処理
    expensive_data = get_expensive_data()
    
    # 手動でキャッシュ操作
    cache.set('user_data', expensive_data, 300)
    cached_data = cache.get('user_data')
    
    return render(request, 'template.html', {'data': cached_data})

永続データ構造の活用

import diskcache as dc

# 永続キュー（Deque）
urls = dc.Deque('web_crawler_urls')
urls.append('https://example.com')
url = urls.popleft() if urls else None

# 永続インデックス（Dict）
results = dc.Index('crawl_results')
results['https://example.com'] = 'page content'

# トランザクション処理
with cache.transact():
    total = cache.incr('total', 123.45)
    count = cache.incr('count')
    
# 統計情報の有効化
cache.stats(enable=True)
# 処理後の統計確認
hits, misses = cache.stats(enable=False, reset=True)
print(f'ヒット率: {hits / (hits + misses) * 100:.2f}%')

非同期処理での活用

import asyncio
import diskcache as dc

cache = dc.Cache()

async def set_async(key, value):
    loop = asyncio.get_running_loop()
    # ThreadPoolExecutorを使用して非同期化
    future = loop.run_in_executor(None, cache.set, key, value)
    result = await future
    return result

async def get_async(key):
    loop = asyncio.get_running_loop()
    future = loop.run_in_executor(None, cache.get, key)
    result = await future
    return result

# 使用例
async def main():
    await set_async('async_key', 'async_value')
    value = await get_async('async_key')
    print(f'取得した値: {value}')

# 実行
asyncio.run(main())