Skip to content

Instantly share code, notes, and snippets.

@bmispelon
Last active June 30, 2024 13:10
Show Gist options
  • Save bmispelon/f064c6859027d6a5767a4f0f3d2ff729 to your computer and use it in GitHub Desktop.
Save bmispelon/f064c6859027d6a5767a4f0f3d2ff729 to your computer and use it in GitHub Desktop.
A Django cache backend to help transition from an old backend to a new one
from django.core.cache import BaseCache, CacheHandler
from django.core.cache.backends.base import DEFAULT_TIMEOUT
from django.core.exceptions import ImproperlyConfigured
class TransitionCache(BaseCache):
"""
A cache backend that helps transitioning from one backend to another.
To use it, set both 'NEW' and 'OLD'as keys in the backend's configuration
dictionary:
CACHES = {
"default": {
"BACKEND": "path.to.TransitionCache",
"OLD": {...}, # the old value of CACHES["default"]
"NEW": {...}, # the future value of CACHES["default"]
}
}
Deploy this as long as necessary (this depends on the cache timeouts you
use in your project), then remove and use onlye the NEW config.
When reading a key, the NEW backend is tried first. If that fails, the OLD
backend is tried second and if that succeeds, the value is written to NEW.
All writes go to NEW.
"""
def __init__(self, location, params):
if "OLD" not in params:
raise ImproperlyConfigured("Missing configuration key OLD for TransitionCache")
if "NEW" not in params:
raise ImproperlyConfigured("Missing configuration key NEW for TransitionCache")
if location:
raise ImproperlyConfigured("Invalid configuration key LOCATION for TransitionCache")
handler = CacheHandler(params)
self.OLD = handler["OLD"]
self.NEW = handler["NEW"]
def add(self, key, value, timeout=DEFAULT_TIMEOUT, version=None):
if self.NEW.add(key, value, timeout=timeout, version=version):
if self.OLD.has_key(key, version=version):
# key was not in NEW, but was in OLD
# so we have to revert the insertion done by NEW.add
self.NEW.delete(key, version=version)
return False
else:
# key was neither in NEW or in OLD, so it was set in NEW
return True
else:
# Key was in NEW, no need to check OLD
return False
def get(self, key, default=None, version=None):
sentinel = object()
value_in_new = self.NEW.get(key, default=sentinel, version=version)
if value_in_new is not sentinel:
return value_in_new
value_in_old = self.OLD.get(key, default=sentinel, version=version)
if value_in_old is not sentinel:
self.NEW.set(key, value_in_old, version=version)
return value_in_old
return default
def set(self, *args, **kwargs):
self.NEW.set(*args, **kwargs)
def touch(self, key, timeout=DEFAULT_TIMEOUT, version=None):
self.make_and_validate_key(key, version=version)
return False
def delete(self, key, version=None):
deleted_in_new = self.NEW.delete(key, version=version)
deleted_in_old = self.OLD.delete(key, version=version)
return deleted_in_new or deleted_in_old
def has_key(self, key, version=None):
has_key_new = self.NEW.has_key(key, version=version)
has_key_old = self.OLD.has_key(key, version=version)
return has_key_new or has_key_old
def clear(self):
self.NEW.clear()
self.OLD.clear()
def close(self):
self.NEW.close()
self.OLD.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment