tessl/pypi-diskcache

Disk Cache -- Disk and file backed persistent cache.

—

Pending

Overview

Eval results

Files

Django Integration

Name: tessl/pypi-diskcache
Author: tessl

DiskCache provides seamless Django integration through the DjangoCache class, which implements Django's cache backend interface while maintaining all DiskCache features and performance benefits. This allows Django applications to use persistent disk-based caching with minimal configuration changes.

Capabilities

DjangoCache Class

A Django-compatible cache backend that wraps FanoutCache while providing the standard Django cache interface.

class DjangoCache:
    def __init__(self, directory, params):
        """
        Initialize Django-compatible cache backend.
        
        Args:
            directory (str): Cache directory path
            params (dict): Django cache configuration parameters:
                - SHARDS (int): Number of shards for FanoutCache. Default 8.
                - DATABASE_TIMEOUT (float): SQLite timeout. Default 0.010.
                - OPTIONS (dict): Additional cache options
        """
        
    @property
    def directory(self):
        """Cache directory path."""

Django Cache Interface

Standard Django cache methods with DiskCache enhancements.

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

def get(self, key, default=None, version=None, read=False, expire_time=False, tag=False, retry=False):
    """
    Retrieve value by key.
    
    Args:
        key (str): Cache key
        default: Default value if key not found
        version (int, optional): Key version for namespacing
        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 on timeout. Default False.
        
    Returns:
        Value, or tuple with additional info if expire_time/tag requested
    """

def set(self, key, value, timeout=DEFAULT_TIMEOUT, version=None, read=False, tag=None, retry=True):
    """
    Store key-value pair.
    
    Args:
        key (str): Cache key
        value: Value to store
        timeout (int): Expiration timeout in seconds
        version (int, optional): Key version for namespacing
        read (bool): Store value as file for reading. Default False.
        tag (str, optional): Tag for grouping related items
        retry (bool): Retry on timeout. Default True.
        
    Returns:
        bool: True if set succeeded
    """

def delete(self, key, version=None, retry=True):
    """
    Delete key from cache.
    
    Args:
        key (str): Cache key to delete
        version (int, optional): Key version for namespacing
        retry (bool): Retry on timeout. Default True.
        
    Returns:
        bool: True if key existed and was deleted
    """

def touch(self, key, timeout=DEFAULT_TIMEOUT, version=None, retry=True):
    """
    Update expiration time for existing key.
    
    Args:
        key (str): Cache key
        timeout (int): New expiration timeout in seconds
        version (int, optional): Key version for namespacing
        retry (bool): Retry on timeout. Default True.
        
    Returns:
        bool: True if key existed and was touched
    """

def incr(self, key, delta=1, version=None, default=None, retry=True):
    """
    Atomically increment numeric value.
    
    Args:
        key (str): Cache key
        delta (int): Amount to increment. Default 1.
        version (int, optional): Key version for namespacing
        default (int, optional): Default value if key doesn't exist
        retry (bool): Retry on timeout. Default True.
        
    Returns:
        New value after increment
        
    Raises:
        ValueError: If key doesn't exist and no default provided
    """

def decr(self, key, delta=1, version=None, default=None, retry=True):
    """
    Atomically decrement numeric value.
    
    Args:
        key (str): Cache key
        delta (int): Amount to decrement. Default 1.
        version (int, optional): Key version for namespacing  
        default (int, optional): Default value if key doesn't exist
        retry (bool): Retry on timeout. Default True.
        
    Returns:
        New value after decrement
        
    Raises:
        ValueError: If key doesn't exist and no default provided
    """

def has_key(self, key, version=None):
    """
    Check if key exists in cache.
    
    Args:
        key (str): Cache key
        version (int, optional): Key version for namespacing
        
    Returns:
        bool: True if key exists
    """

def clear(self):
    """
    Remove all items from cache.
    
    Returns:
        None
    """

DiskCache Extensions

Additional methods that extend Django's cache interface with DiskCache features.

def read(self, key, version=None):
    """
    Get file handle for key stored in read mode.
    
    Args:
        key (str): Cache key
        version (int, optional): Key version for namespacing
        
    Returns:
        File handle or None if key not found
    """

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

Sub-collection Access

Create sub-collections within the Django cache directory.

def cache(self, name):
    """
    Return Cache instance in subdirectory.
    
    Args:
        name (str): Subdirectory name for Cache
        
    Returns:
        Cache: Cache instance in subdirectory
    """

def deque(self, name, maxlen=None):
    """
    Return Deque instance in subdirectory.
    
    Args:
        name (str): Subdirectory name for Deque
        maxlen (int, optional): Maximum length of deque
        
    Returns:
        Deque: Deque instance in subdirectory
    """

def index(self, name):
    """
    Return Index instance in subdirectory.
    
    Args:
        name (str): Subdirectory name for Index
        
    Returns:
        Index: Index instance in subdirectory
    """

Cache Management

Cache management operations with Django compatibility.

def expire(self):
    """
    Remove expired items from cache.
    
    Returns:
        int: Number of expired items removed
    """

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 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."""

def evict(self, tag):
    """
    Remove all items with specified tag.
    
    Args:
        tag (str): Tag to evict
        
    Returns:
        int: Number of items evicted
    """

def cull(self):
    """
    Remove items according to eviction policy.
    
    Returns:
        int: Number of items removed
    """

def close(self, **kwargs):
    """
    Close cache connections and cleanup resources.
    
    Args:
        **kwargs: Additional arguments (for Django compatibility)
    """

Memoization

Django-compatible memoization decorator.

def memoize(self, name=None, timeout=DEFAULT_TIMEOUT, version=None, typed=False, tag=None, ignore=()):
    """
    Memoization decorator using this Django cache.
    
    Args:
        name (str, optional): Name for memoized function. Default function name.
        timeout (int): Expiration timeout in seconds
        version (int, optional): Key version for namespacing
        typed (bool): Distinguish arguments by type. Default False.
        tag (str, optional): Tag for grouping cached results
        ignore (tuple): Argument positions/names to ignore in cache key
        
    Returns:
        Decorator function
    """

Utility Methods

Utility methods for Django integration.

def get_backend_timeout(self, timeout=DEFAULT_TIMEOUT):
    """
    Convert Django timeout format to seconds.
    
    Args:
        timeout (int): Django timeout value
        
    Returns:
        float: Timeout in seconds, or None for no expiration
    """

Django Configuration

Settings Configuration

Configure DiskCache as a Django cache backend in your settings.py:

# settings.py
CACHES = {
    'default': {
        'BACKEND': 'diskcache.djangocache.DjangoCache',
        'LOCATION': '/tmp/django_cache',
        'SHARDS': 8,
        'DATABASE_TIMEOUT': 0.010,
        'OPTIONS': {
            'size_limit': 2**32,  # 4GB
            'eviction_policy': 'least-recently-used',
            'statistics': 1,
            'tag_index': 1,
        },
    },
    'session': {
        'BACKEND': 'diskcache.djangocache.DjangoCache', 
        'LOCATION': '/tmp/django_sessions',
        'SHARDS': 4,
        'OPTIONS': {
            'size_limit': 2**28,  # 256MB
            'eviction_policy': 'least-recently-stored',
        },
    },
}

Multiple Cache Configuration

# Multiple caches for different purposes
CACHES = {
    'default': {
        'BACKEND': 'diskcache.djangocache.DjangoCache',
        'LOCATION': '/var/cache/django/default',
        'SHARDS': 16,
        'OPTIONS': {
            'size_limit': 2**30,  # 1GB
            'eviction_policy': 'least-recently-used',
            'statistics': 1,
        },
    },
    'sessions': {
        'BACKEND': 'diskcache.djangocache.DjangoCache',
        'LOCATION': '/var/cache/django/sessions', 
        'SHARDS': 4,
        'OPTIONS': {
            'size_limit': 2**28,  # 256MB
            'eviction_policy': 'least-recently-stored',
        },
    },
    'database': {
        'BACKEND': 'diskcache.djangocache.DjangoCache',
        'LOCATION': '/var/cache/django/database',
        'SHARDS': 32,
        'OPTIONS': {
            'size_limit': 2**32,  # 4GB
            'eviction_policy': 'least-frequently-used',
            'statistics': 1,
            'tag_index': 1,
        },
    },
}

Usage Examples

Basic Django Cache Usage

from django.core.cache import cache

# Basic cache operations
cache.set('user_count', 1250, timeout=300)  # 5 minutes
user_count = cache.get('user_count', default=0)

# Add only if doesn't exist
cache.add('unique_visitor', True, timeout=86400)  # 24 hours

# Atomic operations
cache.set('page_views', 1000)
cache.incr('page_views')  # Now 1001
cache.decr('page_views', 10)  # Now 991

# Delete operations
cache.delete('temp_data')
cache.clear()  # Remove all items

Django Cache with Versioning

from django.core.cache import cache

# Use versioning for cache invalidation
cache.set('user_profile', user_data, version=1)
profile = cache.get('user_profile', version=1)

# Invalidate by changing version
cache.set('user_profile', updated_data, version=2)
old_profile = cache.get('user_profile', version=1)  # None
new_profile = cache.get('user_profile', version=2)  # updated_data

Using Multiple Caches

from django.core.cache import caches

# Access different cache backends
default_cache = caches['default']
session_cache = caches['sessions'] 
db_cache = caches['database']

# Use each for different purposes
default_cache.set('site_config', config_data)
session_cache.set('user_session', session_data, timeout=1800)
db_cache.set('query_result', query_data, timeout=3600)

DiskCache-specific Features

from django.core.cache import cache

# Use tags for grouped invalidation
cache.set('post_1', post_data, tag='posts', timeout=3600)
cache.set('post_2', other_post, tag='posts', timeout=3600)
cache.set('user_1', user_data, tag='users', timeout=1800)

# Evict all posts
cache.evict('posts')  # Removes post_1 and post_2, keeps user_1

# File-based storage for large items
with open('large_file.pdf', 'rb') as f:
    file_data = f.read()
cache.set('large_document', file_data, read=True)

# Get file handle instead of loading into memory
file_handle = cache.read('large_document')
if file_handle:
    content = file_handle.read()
    file_handle.close()

# Statistics
hits, misses = cache.stats()
print(f"Cache hit ratio: {hits/(hits+misses):.2%}")

Sub-collections in Django

from django.core.cache import cache

# Create specialized data structures
user_cache = cache.cache('users')
task_queue = cache.deque('tasks')  
search_index = cache.index('search')

# Use sub-collections independently
user_cache.set('user:123', user_data)
task_queue.append({'task': 'send_email', 'user_id': 123})
search_index['keyword:python'] = ['doc1', 'doc2', 'doc3']

# These don't interfere with main cache
cache.set('global_setting', 'value')  # Different namespace

Session Backend

# Use DiskCache as Django session backend
# In settings.py:
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'sessions'

# Sessions will be stored in DiskCache automatically
# with all the persistence and performance benefits

Caching Views and Templates

from django.views.decorators.cache import cache_page
from django.core.cache import cache
from django.template.loader import render_to_string

# Cache entire view for 15 minutes
@cache_page(60 * 15, cache='default')
def expensive_view(request):
    # Expensive computation
    data = perform_complex_calculation()
    return render(request, 'template.html', {'data': data})

# Manual template caching
def cached_template_fragment(context_data):
    cache_key = f"template_fragment_{hash(str(context_data))}"
    rendered = cache.get(cache_key)
    
    if rendered is None:
        rendered = render_to_string('fragment.html', context_data)
        cache.set(cache_key, rendered, timeout=3600, tag='templates')
    
    return rendered

# Invalidate template cache when data changes
def invalidate_template_cache():
    cache.evict('templates')  # Remove all cached templates

Database Query Caching

from django.core.cache import cache
from django.db import models
from django.utils.hashlib import md5_constructor

class CachedQueryManager(models.Manager):
    def cached_filter(self, timeout=3600, **kwargs):
        # Create cache key from query parameters
        key_data = str(sorted(kwargs.items()))
        cache_key = f"query_{self.model._meta.label}_{md5_constructor(key_data.encode()).hexdigest()}"
        
        # Try cache first
        result = cache.get(cache_key)
        if result is None:
            result = list(self.filter(**kwargs))
            cache.set(cache_key, result, timeout=timeout, tag=f'model_{self.model._meta.label}')
        
        return result

class Article(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    published = models.BooleanField(default=False)
    
    objects = CachedQueryManager()

# Usage
published_articles = Article.objects.cached_filter(published=True, timeout=1800)

# Invalidate when models change
def invalidate_article_cache():
    cache.evict('model_myapp.Article')

Memoization in Django

from django.core.cache import cache

# Memoize expensive functions
@cache.memoize(timeout=3600, tag='calculations')
def expensive_calculation(x, y, z):
    # Complex computation
    result = perform_expensive_operation(x, y, z)
    return result

# Memoize with versioning
@cache.memoize(name='user_permissions', timeout=1800, version=1)
def get_user_permissions(user_id):
    # Database queries to get permissions
    return User.objects.get(id=user_id).get_all_permissions()

# Use in views
def user_dashboard(request):
    permissions = get_user_permissions(request.user.id)
    calculation_result = expensive_calculation(1, 2, 3)
    
    return render(request, 'dashboard.html', {
        'permissions': permissions,
        'result': calculation_result
    })

# Invalidate memoized functions
def invalidate_calculations():
    cache.evict('calculations')

Best Practices

Cache Key Management

# Use consistent key naming
CACHE_KEY_PATTERNS = {
    'user_profile': 'user:profile:{user_id}',
    'article_list': 'articles:list:{category}:{page}',
    'search_results': 'search:{query_hash}',
}

def get_cache_key(pattern, **kwargs):
    return CACHE_KEY_PATTERNS[pattern].format(**kwargs)

# Usage
cache_key = get_cache_key('user_profile', user_id=123)
cache.set(cache_key, user_data)

Timeout Strategies

# Different timeouts for different data types
CACHE_TIMEOUTS = {
    'static_content': 86400,    # 24 hours
    'user_sessions': 1800,      # 30 minutes  
    'api_responses': 300,       # 5 minutes
    'database_queries': 3600,   # 1 hour
    'template_fragments': 7200, # 2 hours
}

cache.set('config', data, timeout=CACHE_TIMEOUTS['static_content'])

Error Handling

from django.core.cache import cache
import logging

def safe_cache_get(key, default=None):
    try:
        return cache.get(key, default)
    except Exception as e:
        logging.error(f"Cache get error for key {key}: {e}")
        return default

def safe_cache_set(key, value, timeout=300):
    try:
        return cache.set(key, value, timeout=timeout)
    except Exception as e:
        logging.error(f"Cache set error for key {key}: {e}")
        return False

Monitoring and Maintenance

# Management command for cache monitoring
from django.core.management.base import BaseCommand
from django.core.cache import cache

class Command(BaseCommand):
    help = 'Monitor cache statistics'
    
    def handle(self, *args, **options):
        hits, misses = cache.stats()
        total = hits + misses
        
        if total > 0:
            hit_ratio = hits / total
            self.stdout.write(f"Cache Statistics:")
            self.stdout.write(f"  Hits: {hits}")
            self.stdout.write(f"  Misses: {misses}")
            self.stdout.write(f"  Hit Ratio: {hit_ratio:.2%}")
        
        # Cleanup expired items
        expired = cache.expire()
        self.stdout.write(f"Expired {expired} items")
        
        # Show cache size
        volume = cache.volume() if hasattr(cache, 'volume') else 'Unknown'
        self.stdout.write(f"Cache Size: {volume} bytes")

Install with Tessl CLI