tessl/pypi-diskcache
Disk Cache -- Disk and file backed persistent cache.
—
Pending
persistent-data-structures.mddocs/
Persistent Data Structures
DiskCache provides persistent, disk-backed data structures that offer familiar interfaces while storing data reliably on disk. The Deque class implements a double-ended queue with the collections.abc.Sequence interface, while the Index class implements a mapping/dictionary with the collections.abc.MutableMapping interface.
Capabilities
Deque - Persistent Double-Ended Queue
A persistent sequence supporting efficient append and pop operations from both ends, with full compatibility with Python's collections.abc.Sequence interface.
class Deque:
def __init__(self, iterable=(), directory=None, maxlen=None):
"""
Initialize persistent deque.
Args:
iterable: Initial values to populate deque
directory (str, optional): Directory for storage. If None, creates temp directory.
maxlen (int, optional): Maximum length. When full, adding items removes from opposite end.
"""
@classmethod
def fromcache(cls, cache, iterable=(), maxlen=None):
"""
Create Deque from existing Cache instance.
Args:
cache (Cache): Existing Cache instance to use for storage
iterable: Initial values to populate deque
maxlen (int, optional): Maximum length
Returns:
Deque: New Deque instance using the provided cache
"""
@property
def cache(self):
"""Underlying Cache instance used for storage."""
@property
def directory(self):
"""Directory path for storage."""
@property
def maxlen(self):
"""Maximum length (None if unlimited)."""
@maxlen.setter
def maxlen(self, value):
"""Set maximum length."""Sequence Operations
Standard sequence operations with persistent storage.
def __getitem__(self, index):
"""
Get item by index using deque[index] syntax.
Args:
index (int): Index (supports negative indexing)
Returns:
Item at the specified index
Raises:
IndexError: If index is out of range
"""
def __setitem__(self, index, value):
"""
Set item by index using deque[index] = value syntax.
Args:
index (int): Index (supports negative indexing)
value: Value to set
Raises:
IndexError: If index is out of range
"""
def __delitem__(self, index):
"""
Delete item by index using del deque[index] syntax.
Args:
index (int): Index (supports negative indexing)
Raises:
IndexError: If index is out of range
"""
def __len__(self):
"""Get number of items in deque."""
def __iter__(self):
"""Iterate from front to back."""
def __reversed__(self):
"""Iterate from back to front."""
def __contains__(self, value):
"""Check if value exists in deque using 'value in deque' syntax."""Deque-Specific Operations
Efficient operations for double-ended queue functionality.
def append(self, value):
"""
Add value to the back (right side) of deque.
Args:
value: Value to append
"""
def appendleft(self, value):
"""
Add value to the front (left side) of deque.
Args:
value: Value to append to front
"""
def pop(self):
"""
Remove and return value from the back (right side) of deque.
Returns:
Value from back of deque
Raises:
IndexError: If deque is empty
"""
def popleft(self):
"""
Remove and return value from the front (left side) of deque.
Returns:
Value from front of deque
Raises:
IndexError: If deque is empty
"""
def peek(self):
"""
Return value from back without removing it.
Returns:
Value from back of deque
Raises:
IndexError: If deque is empty
"""
def peekleft(self):
"""
Return value from front without removing it.
Returns:
Value from front of deque
Raises:
IndexError: If deque is empty
"""Extended Operations
Additional operations for working with sequences and managing the deque.
def extend(self, iterable):
"""
Extend deque by appending elements from iterable to the back.
Args:
iterable: Sequence of values to append
"""
def extendleft(self, iterable):
"""
Extend deque by appending elements from iterable to the front.
Note: Order is reversed - first item in iterable becomes last added.
Args:
iterable: Sequence of values to append to front
"""
def clear(self):
"""Remove all elements from deque."""
def copy(self):
"""
Create shallow copy of deque.
Returns:
Deque: New deque with same elements
"""
def count(self, value):
"""
Count occurrences of value in deque.
Args:
value: Value to count
Returns:
int: Number of occurrences
"""
def remove(self, value):
"""
Remove first occurrence of value from deque.
Args:
value: Value to remove
Raises:
ValueError: If value is not found
"""
def reverse(self):
"""Reverse the deque in place."""
def rotate(self, steps=1):
"""
Rotate deque steps positions to the right.
Args:
steps (int): Number of steps to rotate. Positive values rotate right,
negative values rotate left. Default 1.
"""Comparison Operations
Rich comparison operations following sequence semantics.
def __eq__(self, other):
"""
Test equality with another sequence.
Args:
other: Another sequence to compare
Returns:
bool: True if sequences are equal
"""
def __ne__(self, other):
"""Test inequality with another sequence."""
def __lt__(self, other):
"""Test if lexicographically less than another sequence."""
def __le__(self, other):
"""Test if lexicographically less than or equal to another sequence."""
def __gt__(self, other):
"""Test if lexicographically greater than another sequence."""
def __ge__(self, other):
"""Test if lexicographically greater than or equal to another sequence."""Advanced Deque Operations
Advanced operations for transactions and serialization.
def __iadd__(self, iterable):
"""
In-place concatenation using deque += iterable syntax.
Args:
iterable: Sequence of values to append
Returns:
Deque: Self (for chaining)
"""
def __repr__(self):
"""String representation of deque."""
def transact(self):
"""
Context manager for atomic transactions.
Returns:
Context manager for transaction on underlying cache
"""
def __getstate__(self):
"""Support for pickle serialization."""
def __setstate__(self, state):
"""Support for pickle deserialization."""Index - Persistent Mapping
A persistent mapping/dictionary that implements the collections.abc.MutableMapping interface with disk-based storage.
class Index:
def __init__(self, *args, **kwargs):
"""
Initialize persistent mapping.
Args:
directory (str, optional): Directory for storage as first positional arg
Other args/kwargs: Initial mapping data (same as dict constructor)
"""
@classmethod
def fromcache(cls, cache, *args, **kwargs):
"""
Create Index from existing Cache instance.
Args:
cache (Cache): Existing Cache instance to use for storage
Other args/kwargs: Initial mapping data
Returns:
Index: New Index instance using the provided cache
"""
@property
def cache(self):
"""Underlying Cache instance used for storage."""
@property
def directory(self):
"""Directory path for storage."""Mapping Operations
Standard mapping operations with persistent storage.
def __getitem__(self, key):
"""
Get value by key using index[key] syntax.
Args:
key: Key to retrieve
Returns:
Value associated with key
Raises:
KeyError: If key is not found
"""
def __setitem__(self, key, value):
"""
Set key-value pair using index[key] = value syntax.
Args:
key: Key to store
value: Value to associate with key
"""
def __delitem__(self, key):
"""
Delete key using del index[key] syntax.
Args:
key: Key to delete
Raises:
KeyError: If key is not found
"""
def __len__(self):
"""Get number of key-value pairs in index."""
def __iter__(self):
"""Iterate over keys in sorted order."""
def __reversed__(self):
"""Iterate over keys in reverse sorted order."""
def __contains__(self, key):
"""Check if key exists using 'key in index' syntax."""Dictionary Methods
Standard dictionary methods for persistent mapping.
def get(self, key, default=None):
"""
Get value with default fallback.
Args:
key: Key to retrieve
default: Default value if key not found
Returns:
Value associated with key or default
"""
def pop(self, key, default=ENOVAL):
"""
Remove key and return its value.
Args:
key: Key to remove
default: Default value if key not found (optional)
Returns:
Value that was associated with key
Raises:
KeyError: If key not found and no default provided
"""
def popitem(self, last=True):
"""
Remove and return arbitrary key-value pair.
Args:
last (bool): If True, LIFO order; if False, FIFO order. Default True.
Returns:
Tuple of (key, value)
Raises:
KeyError: If index is empty
"""
def setdefault(self, key, default=None):
"""
Get value for key, setting to default if key doesn't exist.
Args:
key: Key to get or set
default: Default value to set if key doesn't exist
Returns:
Existing value for key or default (which is also stored)
"""
def clear(self):
"""Remove all key-value pairs from index."""
def update(self, *args, **kwargs):
"""
Update index with key-value pairs from other mapping or iterable.
Args:
Same as dict.update() - mapping, iterable of pairs, or keyword args
"""View Objects
Dictionary view objects for keys, values, and items.
def keys(self):
"""
Return KeysView of index keys.
Returns:
KeysView: View of keys that updates with index changes
"""
def values(self):
"""
Return ValuesView of index values.
Returns:
ValuesView: View of values that updates with index changes
"""
def items(self):
"""
Return ItemsView of index key-value pairs.
Returns:
ItemsView: View of (key, value) pairs that updates with index changes
"""Queue Operations
Use the index as a queue with key-value pairs.
def push(self, value, prefix=None, side='back'):
"""
Push value to queue with generated key.
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'.
Returns:
Generated key for the pushed value
"""
def pull(self, prefix=None, default=(None, None), side='front'):
"""
Pull key-value pair 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'.
Returns:
Tuple of (key, value) or default if queue empty
"""
def peekitem(self, last=True):
"""
Peek at key-value pair without removing it.
Args:
last (bool): Peek at last item if True, first if False. Default True.
Returns:
Tuple of (key, value) or (None, None) if index empty
"""Comparison Operations
Comparison operations for mappings.
def __eq__(self, other):
"""
Test equality with another mapping.
Args:
other: Another mapping to compare
Returns:
bool: True if mappings have same key-value pairs
"""
def __ne__(self, other):
"""Test inequality with another mapping."""Advanced Index Operations
Advanced operations for memoization, transactions, and serialization.
def __repr__(self):
"""String representation of index."""
def memoize(self, name=None, typed=False, ignore=()):
"""
Memoization decorator using this index.
Args:
name (str, optional): Name for memoized function. Default function name.
typed (bool): Distinguish arguments by type. Default False.
ignore (tuple): Argument positions/names to ignore in caching
Returns:
Decorator function
"""
def transact(self):
"""
Context manager for atomic transactions.
Returns:
Context manager for transaction on underlying cache
"""
def __getstate__(self):
"""Support for pickle serialization."""
def __setstate__(self, state):
"""Support for pickle deserialization."""Usage Examples
Deque Usage
import diskcache
# Create persistent deque
deque = diskcache.Deque('/tmp/mydeque')
# Basic deque operations
deque.append('item1')
deque.appendleft('item0')
deque.extend(['item2', 'item3'])
print(f"Length: {len(deque)}") # 4
print(f"Front: {deque.peekleft()}") # 'item0'
print(f"Back: {deque.peek()}") # 'item3'
# Pop items
back_item = deque.pop() # 'item3'
front_item = deque.popleft() # 'item0'
# Sequence operations
print(deque[0]) # 'item1'
deque[0] = 'modified_item1'
# Bounded deque
bounded = diskcache.Deque(maxlen=3)
bounded.extend([1, 2, 3, 4, 5]) # Only keeps last 3: [3, 4, 5]Index Usage
import diskcache
# Create persistent index
index = diskcache.Index('/tmp/myindex')
# Basic mapping operations
index['user:123'] = {'name': 'Alice', 'age': 30}
index['user:456'] = {'name': 'Bob', 'age': 25}
print(index['user:123']) # {'name': 'Alice', 'age': 30}
print(len(index)) # 2
# Dictionary methods
user = index.get('user:789', {'name': 'Unknown'})
index.setdefault('settings', {'theme': 'light'})
# Iterate over keys, values, items
for key in index:
print(f"Key: {key}")
for value in index.values():
print(f"Value: {value}")
for key, value in index.items():
print(f"{key}: {value}")
# Queue operations with generated keys
task_key = index.push({'task': 'send_email', 'user_id': 123})
key, task = index.pull()Advanced Usage
# Create from existing cache
cache = diskcache.Cache('/tmp/shared')
deque = diskcache.Deque.fromcache(cache, [1, 2, 3])
index = diskcache.Index.fromcache(cache, {'a': 1, 'b': 2})
# Atomic operations
with index.transact():
index['counter'] = index.get('counter', 0) + 1
index['last_update'] = time.time()
# Memoization with Index
@index.memoize(typed=True)
def expensive_function(x, y):
return x ** y
result = expensive_function(2, 10) # Computed and stored
result = expensive_function(2, 10) # Retrieved from index
# Deque as a work queue
work_queue = diskcache.Deque('/tmp/work_queue')
# Producer
for i in range(100):
work_queue.append({'job_id': i, 'data': f'job_{i}'})
# Consumer
while len(work_queue) > 0:
job = work_queue.popleft()
print(f"Processing {job}")Performance Considerations
# For large datasets, consider using transactions
large_index = diskcache.Index('/tmp/large_data')
# Bulk insert with transaction
with large_index.transact():
for i in range(10000):
large_index[f'key_{i}'] = {'value': i, 'squared': i**2}
# Bounded deque for fixed-size buffers
log_buffer = diskcache.Deque('/tmp/logs', maxlen=1000)
# Always keeps only the last 1000 log entries
for i in range(5000):
log_buffer.append(f'Log entry {i}')
print(len(log_buffer)) # 1000 (not 5000)Install with Tessl CLI
npx tessl i tessl/pypi-diskcache