Disk Cache -- Disk and file backed persistent cache.
npx @tessl/cli install tessl/pypi-diskcache@5.6.0DiskCache is a comprehensive disk-based caching library for Python that provides high-performance persistent storage using SQLite databases and memory-mapped files. It offers thread-safe and process-safe caching with multiple eviction policies (LRU, LFU), supports various data structures including Cache, FanoutCache, Deque, and Index, and provides Django integration. The library is designed for applications requiring gigabytes of cache storage with performance that matches or exceeds in-memory solutions like Memcached, while being completely written in pure Python with no external dependencies.
pip install diskcacheimport diskcacheMain classes:
from diskcache import Cache, FanoutCache, Deque, IndexConstants and settings:
from diskcache import DEFAULT_SETTINGS, EVICTION_POLICY, ENOVAL, UNKNOWNRecipe classes for synchronization:
from diskcache import Lock, RLock, BoundedSemaphore, AveragerRecipe functions:
from diskcache import throttle, barrier, memoize_stampedeDjango integration (if Django is available):
from diskcache import DjangoCacheimport diskcache
# Create a simple cache
cache = diskcache.Cache('/tmp/mycache')
# Store and retrieve values
cache.set('key', 'value')
value = cache.get('key')
# Dict-like interface
cache['another_key'] = {'data': [1, 2, 3]}
data = cache['another_key']
# Context manager for transactions
with cache.transact():
cache.set('a', 1)
cache.set('b', 2)
# Use as a decorator for memoization
@cache.memoize()
def expensive_function(x, y):
# ... expensive computation
return x * y
# Clean up
cache.close()DiskCache provides multiple data structure classes built on a common foundation:
All classes support context management, atomic transactions, and provide comprehensive cache management with statistics, eviction policies, and tag-based operations.
Primary caching functionality with the Cache class, supporting all standard cache operations, atomic transactions, queue operations, and comprehensive cache management with statistics and eviction policies.
class Cache:
def __init__(self, directory=None, timeout=60, disk=Disk, **settings): ...
def set(self, key, value, expire=None, read=False, tag=None, retry=False): ...
def get(self, key, default=None, read=False, expire_time=False, tag=False, retry=False): ...
def delete(self, key, retry=False): ...
def clear(self, retry=False): ...
def close(self): ...High-throughput caching using FanoutCache which automatically distributes keys across multiple Cache instances for improved performance and scalability.
class FanoutCache:
def __init__(self, directory=None, shards=8, timeout=0.010, disk=Disk, **settings): ...
def set(self, key, value, expire=None, read=False, tag=None, retry=False): ...
def get(self, key, default=None, read=False, expire_time=False, tag=False, retry=False): ...
def cache(self, name, timeout=60, disk=None, **settings): ...
def deque(self, name, maxlen=None): ...
def index(self, name): ...Disk-backed data structures (Deque and Index) that provide familiar interfaces while persisting data to disk, supporting full sequence and mapping operations with atomic transactions.
class Deque:
def __init__(self, iterable=(), directory=None, maxlen=None): ...
def append(self, value): ...
def appendleft(self, value): ...
def pop(self): ...
def popleft(self): ...
class Index:
def __init__(self, *args, **kwargs): ...
def __getitem__(self, key): ...
def __setitem__(self, key, value): ...
def keys(self): ...
def values(self): ...
def items(self): ...Thread-safe and process-safe synchronization primitives including locks, semaphores, averager, and various decorators for throttling and cache stampede protection.
class Lock:
def __init__(self, cache, key, expire=None, tag=None): ...
def acquire(self): ...
def release(self): ...
class RLock:
def __init__(self, cache, key, expire=None, tag=None): ...
def acquire(self): ...
def release(self): ...
class BoundedSemaphore:
def __init__(self, cache, key, value=1, expire=None, tag=None): ...
def acquire(self): ...
def release(self): ...
class Averager:
def __init__(self, cache, key, expire=None, tag=None): ...
def add(self, value): ...
def get(self): ...
def pop(self): ...Decorator functions for advanced caching patterns including throttling, serialization barriers, and memoization with cache stampede protection.
def throttle(cache, count, seconds, name=None, expire=None, tag=None,
time_func=time.time, sleep_func=time.sleep): ...
def barrier(cache, lock_factory, name=None, expire=None, tag=None): ...
def memoize_stampede(cache, expire, name=None, typed=False, tag=None,
beta=1, ignore=()): ...Serialization engines that handle conversion between Python objects and disk storage, supporting multiple storage modes including pickle, JSON with compression, and raw binary data.
class Disk:
def __init__(self, directory, min_file_size=0, pickle_protocol=0): ...
def put(self, key): ...
def get(self, key, raw): ...
def store(self, value, read, key=UNKNOWN): ...
def fetch(self, mode, filename, value, read): ...
class JSONDisk:
def __init__(self, directory, compress_level=1, **kwargs): ...
# Inherits all Disk methods with JSON serializationFull Django cache backend compatibility providing seamless integration with Django applications while maintaining all DiskCache features and performance benefits.
class DjangoCache:
def __init__(self, directory, params): ...
def add(self, key, value, timeout=DEFAULT_TIMEOUT, version=None): ...
def get(self, key, default=None, version=None): ...
def set(self, key, value, timeout=DEFAULT_TIMEOUT, version=None): ...
def delete(self, key, version=None): ...
def clear(self): ...DEFAULT_SETTINGS = {
'statistics': 0,
'tag_index': 0,
'eviction_policy': 'least-recently-stored',
'size_limit': 2**30, # 1GB
'cull_limit': 10,
'sqlite_auto_vacuum': 1,
'sqlite_cache_size': 2**13,
'sqlite_journal_mode': 'wal',
'sqlite_mmap_size': 2**26,
'sqlite_synchronous': 1,
'disk_min_file_size': 2**15,
'disk_pickle_protocol': pickle.HIGHEST_PROTOCOL,
}
EVICTION_POLICY = {
'none': {...},
'least-recently-stored': {...},
'least-recently-used': {...},
'least-frequently-used': {...},
}
ENOVAL = Constant('ENOVAL') # Special constant for "no value"
UNKNOWN = Constant('UNKNOWN') # Special constant for unknown valuesclass Timeout(Exception):
"""Database timeout exception."""
class UnknownFileWarning(UserWarning):
"""Warning for unknown files in cache directory."""
class EmptyDirWarning(UserWarning):
"""Warning for empty cache directories."""