tessl/pypi-diskcache

Disk Cache -- Disk and file backed persistent cache.

—

Pending

Overview

Eval results

Files

Core Caching

Name: tessl/pypi-diskcache
Author: tessl

The Cache class provides the primary caching functionality in DiskCache, offering thread-safe and process-safe disk-based storage with SQLite as the backend. It supports all standard cache operations, atomic transactions, queue operations, and comprehensive cache management with statistics and eviction policies.

Capabilities

Cache Initialization

Create a Cache instance with optional directory path, timeout settings, and custom serialization.

class Cache:
    def __init__(self, directory=None, timeout=60, disk=Disk, **settings):
        """
        Initialize cache instance.
        
        Args:
            directory (str, optional): Cache directory path. If None, creates temp directory.
            timeout (float): SQLite connection timeout in seconds. Default 60.
            disk (Disk): Disk instance for serialization. Default Disk.
            **settings: Cache configuration options from DEFAULT_SETTINGS.
        """

    @property
    def directory(self):
        """Cache directory path."""
        
    @property
    def timeout(self):
        """SQLite connection timeout."""
        
    @property
    def disk(self):
        """Disk instance used for serialization."""

Basic Cache Operations

Store, retrieve, and delete key-value pairs with optional expiration, read mode, and tagging.

def set(self, key, value, expire=None, read=False, tag=None, retry=False):
    """
    Store key-value pair in cache.
    
    Args:
        key: Cache key (must be hashable)
        value: Value to store
        expire (float, optional): Expiration time in seconds from now
        read (bool): Store value as file for reading. Default False.
        tag (str, optional): Tag for grouping related items
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        bool: True if set succeeded
    """

def get(self, key, default=None, read=False, expire_time=False, tag=False, retry=False):
    """
    Retrieve value by key.
    
    Args:
        key: Cache key
        default: Default value if key not found
        read (bool): Return file handle instead of value. Default False.
        expire_time (bool): Include expiration time in result. Default False.
        tag (bool): Include tag in result. Default False.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        Value, or tuple with additional info if expire_time/tag requested
    """

def delete(self, key, retry=False):
    """
    Delete key from cache.
    
    Args:
        key: Cache key to delete
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        bool: True if key existed and was deleted
    """

def add(self, key, value, expire=None, read=False, tag=None, retry=False):
    """
    Add key-value pair only if key doesn't exist.
    
    Args:
        key: Cache key
        value: Value to store
        expire (float, optional): Expiration time in seconds from now
        read (bool): Store value as file for reading. Default False.
        tag (str, optional): Tag for grouping related items
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        bool: True if key was added (didn't exist)
    """

def touch(self, key, expire=None, retry=False):
    """
    Update expiration time for existing key.
    
    Args:
        key: Cache key
        expire (float, optional): New expiration time in seconds from now
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        bool: True if key existed and was touched
    """

def pop(self, key, default=None, expire_time=False, tag=False, retry=False):
    """
    Remove and return value for key.
    
    Args:
        key: Cache key
        default: Default value if key not found
        expire_time (bool): Include expiration time in result. Default False.
        tag (bool): Include tag in result. Default False.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        Value, or tuple with additional info if expire_time/tag requested
    """

def read(self, key, retry=False):
    """
    Get file handle for key stored in read mode.
    
    Args:
        key: Cache key
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        File handle or None if key not found
    """

Dict-like Interface

Use familiar dictionary syntax for cache operations.

def __setitem__(self, key, value):
    """Store key-value pair using cache[key] = value syntax."""

def __getitem__(self, key):
    """Retrieve value using cache[key] syntax. Raises KeyError if not found."""

def __delitem__(self, key, retry=True):
    """Delete key using del cache[key] syntax. Raises KeyError if not found."""

def __contains__(self, key):
    """Check if key exists using 'key in cache' syntax."""

def __len__(self):
    """Get count of items in cache."""

def __iter__(self):
    """Iterate over cache keys."""

def __reversed__(self):
    """Reverse iterate over cache keys."""

Atomic Operations

Perform atomic increment and decrement operations.

def incr(self, key, delta=1, default=0, retry=False):
    """
    Atomically increment numeric value.
    
    Args:
        key: Cache key
        delta (int): Amount to increment. Default 1.
        default (int): Default value if key doesn't exist. Default 0.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        New value after increment
    """

def decr(self, key, delta=1, default=0, retry=False):
    """
    Atomically decrement numeric value.
    
    Args:
        key: Cache key  
        delta (int): Amount to decrement. Default 1.
        default (int): Default value if key doesn't exist. Default 0.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        New value after decrement
    """

Queue Operations

Use the cache as a queue with push/pull operations supporting prefixes and different sides.

def push(self, value, prefix=None, side='back', expire=None, read=False, tag=None, retry=False):
    """
    Push value to queue.
    
    Args:
        value: Value to push
        prefix (str, optional): Key prefix for queue namespace
        side (str): Which side to push to ('back' or 'front'). Default 'back'.
        expire (float, optional): Expiration time in seconds from now
        read (bool): Store value as file for reading. Default False.
        tag (str, optional): Tag for grouping related items
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        Generated key for the pushed value
    """

def pull(self, prefix=None, default=(None, None), side='front', expire_time=False, tag=False, retry=False):
    """
    Pull value from queue.
    
    Args:
        prefix (str, optional): Key prefix for queue namespace
        default: Default value if queue is empty. Default (None, None).
        side (str): Which side to pull from ('front' or 'back'). Default 'front'.
        expire_time (bool): Include expiration time in result. Default False.
        tag (bool): Include tag in result. Default False.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        Tuple of (key, value) or default if queue empty
    """

def peek(self, prefix=None, default=(None, None), side='front', expire_time=False, tag=False, retry=False):
    """
    Peek at queue value without removing it.
    
    Args:
        prefix (str, optional): Key prefix for queue namespace
        default: Default value if queue is empty. Default (None, None).
        side (str): Which side to peek at ('front' or 'back'). Default 'front'.
        expire_time (bool): Include expiration time in result. Default False.
        tag (bool): Include tag in result. Default False.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        Tuple of (key, value) or default if queue empty
    """

def peekitem(self, last=True, expire_time=False, tag=False, retry=False):
    """
    Peek at any item in cache without removing it.
    
    Args:
        last (bool): Peek at last item if True, first if False. Default True.
        expire_time (bool): Include expiration time in result. Default False.
        tag (bool): Include tag in result. Default False.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        Tuple of (key, value) or (None, None) if cache empty
    """

Cache Management

Manage cache contents with clearing, culling, expiration, and eviction operations.

def clear(self, retry=False):
    """
    Remove all items from cache.
    
    Args:
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        int: Number of items removed
    """

def cull(self, retry=False):
    """
    Remove items according to eviction policy.
    
    Args:
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        int: Number of items removed
    """

def expire(self, now=None, retry=False):
    """
    Remove expired items from cache.
    
    Args:
        now (float, optional): Current time. Default time.time().
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        int: Number of expired items removed
    """

def evict(self, tag, retry=False):
    """
    Remove all items with specified tag.
    
    Args:
        tag (str): Tag to evict
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        int: Number of items evicted
    """

def check(self, fix=False, retry=False):
    """
    Check cache database consistency.
    
    Args:
        fix (bool): Attempt to fix issues if found. Default False.
        retry (bool): Retry operation on timeout. Default False.
        
    Returns:
        List of issues found (empty if no issues)
    """

Tag Management

Create and manage database indexes for tag-based operations.

def create_tag_index(self):
    """Create database index on tag column for faster tag operations."""

def drop_tag_index(self):
    """Drop database index on tag column."""

Statistics and Iteration

Access cache statistics and iterate through cache contents.

def stats(self, enable=True, reset=False):
    """
    Get cache hit/miss statistics.
    
    Args:
        enable (bool): Enable statistics tracking. Default True.
        reset (bool): Reset statistics counters. Default False.
        
    Returns:
        Tuple of (hits, misses) counters
    """

def volume(self):
    """
    Get total cache size on disk.
    
    Returns:
        int: Total size in bytes
    """

def iterkeys(self, reverse=False):
    """
    Iterate cache keys in sorted order.
    
    Args:
        reverse (bool): Iterate in reverse order. Default False.
        
    Yields:
        Cache keys in sorted order
    """

Transaction Management

Use context managers and transactions for atomic operations.

def transact(self, retry=False):
    """
    Context manager for atomic transactions.
    
    Args:
        retry (bool): Retry transaction on timeout. Default False.
        
    Returns:
        Context manager for transaction
    """

def __enter__(self):
    """Context manager entry - opens database connection."""

def __exit__(self, *exception):
    """Context manager exit - closes database connection."""

def close(self):
    """Close database connection and cleanup resources."""

Memoization Decorator

Use the cache as a memoization decorator for functions.

def memoize(self, name=None, typed=False, expire=None, tag=None, ignore=()):
    """
    Memoization decorator using this cache.
    
    Args:
        name (str, optional): Name for memoized function. Default function name.
        typed (bool): Distinguish arguments by type. Default False.
        expire (float, optional): Expiration time in seconds from now
        tag (str, optional): Tag for grouping related items
        ignore (tuple): Argument positions/names to ignore in caching
        
    Returns:
        Decorator function
    """

Advanced Operations

Advanced cache operations for settings management and serialization support.

def reset(self, key, value=ENOVAL, update=True):
    """
    Reset cache setting value.
    
    Args:
        key (str): Setting key from DEFAULT_SETTINGS
        value: New value for setting
        update (bool): Update existing cache with new setting. Default True.
        
    Returns:
        Previous value of setting
    """

def __getstate__(self):
    """Support for pickle serialization - returns cache state."""

def __setstate__(self, state):
    """Support for pickle deserialization - restores cache state."""

Usage Examples

Basic Usage

import diskcache

# Create cache
cache = diskcache.Cache('/tmp/mycache')

# Basic operations
cache.set('user:123', {'name': 'Alice', 'age': 30})
user = cache.get('user:123')

# Dict-like interface
cache['settings'] = {'theme': 'dark', 'lang': 'en'}
if 'settings' in cache:
    settings = cache['settings']

# Expiration
cache.set('temp_data', 'expires in 1 hour', expire=3600)

# Atomic operations
cache.set('counter', 0)
cache.incr('counter')  # Returns 1
cache.decr('counter')  # Returns 0

Transaction Example

# Atomic operations
with cache.transact():
    cache.set('a', 1)
    cache.set('b', 2)
    cache.incr('counter')
    # All operations committed together

Queue Usage

# Use as a queue
task_id = cache.push({'task': 'process_data', 'priority': 1}, prefix='tasks')
key, task = cache.pull(prefix='tasks')

# LIFO queue (stack)
cache.push('item1', side='back')
cache.push('item2', side='back')
key, item = cache.pull(side='back')  # Gets 'item2'

Memoization

@cache.memoize(expire=3600)  # Cache for 1 hour
def expensive_computation(n):
    # Expensive operation
    return sum(i**2 for i in range(n))

result = expensive_computation(1000)  # Computed and cached
result = expensive_computation(1000)  # Retrieved from cache

Install with Tessl CLI