Last active
June 30, 2024 13:10
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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