"""
Pluggable cache backend abstraction.
Provides:
- CacheBackend: abstract interface for persistent cache storage
- DiskBackend: file-based persistence using pickle (stdlib only)
Designed for extensibility: Redis, Memcached, or custom backends can be
added by subclassing CacheBackend without changing core cache logic.
"""
import os
import pickle
import threading
import time as _time
from abc import ABC, abstractmethod
[docs]
class CacheBackend(ABC):
"""Abstract interface for persistent cache storage.
Subclass this to implement Redis, Memcached, or any other backend.
All methods are synchronous; async wrappers live in the cache layer.
"""
[docs]
@abstractmethod
def load(self):
"""Load persisted data and return a dict of {key: (value, expiration)}.
expiration is a float (monotonic deadline) or None (no TTL).
Expired entries MAY be included; the caller filters them.
Returns an empty dict if no persisted data exists.
"""
[docs]
@abstractmethod
def save(self, data):
"""Persist cache data.
:param data: dict of {key: (value, ttl_remaining_seconds_or_None)}
ttl_remaining is seconds remaining (float) or None for no-TTL entries.
"""
[docs]
@abstractmethod
def clear(self):
"""Remove all persisted data."""
[docs]
class DiskBackend(CacheBackend):
"""File-based cache persistence using pickle.
Data is written atomically (write-to-temp then rename) to prevent
corruption on crash. No external dependencies required.
:param path: filesystem path for the cache file (e.g. '/tmp/my_cache.pkl')
"""
def __init__(self, path):
self.path = path
self._lock = threading.Lock()
[docs]
def load(self):
"""Load cache entries from disk.
Returns {key: (value, monotonic_expiration_or_None)}.
Translates stored relative TTLs back to absolute monotonic deadlines,
subtracting time elapsed since save.
"""
with self._lock:
if not os.path.exists(self.path):
return {}
try:
with open(self.path, 'rb') as f:
payload = pickle.load(f)
except (OSError, pickle.UnpicklingError, EOFError, ValueError):
return {}
# payload format: {'save_time': float, 'entries': {key: (value, ttl_remaining)}}
# Legacy format (pre-timestamp): plain dict {key: (value, ttl_remaining)}
if isinstance(payload, dict) and 'save_time' in payload and 'entries' in payload:
save_time = payload['save_time']
raw = payload['entries']
else:
# Legacy format: no timestamp, treat ttl_remaining as-is
save_time = _time.time()
raw = payload
elapsed = _time.time() - save_time
now = _time.monotonic()
result = {}
for key, (value, ttl_remaining) in raw.items():
if ttl_remaining is not None:
adjusted = ttl_remaining - elapsed
if adjusted <= 0:
continue # expired since save
expiration = now + adjusted
else:
expiration = None
result[key] = (value, expiration)
return result
[docs]
def save(self, data):
"""Persist cache entries to disk atomically.
:param data: {key: (value, ttl_remaining_seconds_or_None)}
"""
payload = {
'save_time': _time.time(),
'entries': data,
}
with self._lock:
tmp_path = self.path + '.tmp'
try:
parent = os.path.dirname(self.path)
if parent:
os.makedirs(parent, exist_ok=True)
with open(tmp_path, 'wb') as f:
pickle.dump(payload, f, protocol=pickle.HIGHEST_PROTOCOL)
os.replace(tmp_path, self.path)
except OSError:
# Best-effort; don't crash the application on save failure
try:
os.unlink(tmp_path)
except OSError:
pass
[docs]
def clear(self):
"""Remove the persisted cache file."""
with self._lock:
try:
os.unlink(self.path)
except OSError:
pass