Skip to content

Instantly share code, notes, and snippets.

@bkonkle
Created June 3, 2010 23:00
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save bkonkle/424637 to your computer and use it in GitHub Desktop.
Save bkonkle/424637 to your computer and use it in GitHub Desktop.
Jeremy Dunck wrote this cache backend for Pegasus News, and I've been given permission to open-source it. I'm going to create a full Github repository with a setup.py for easy pip install, and I also plan to contribute it to Django. For now though, I'm
import types
from django.conf import settings
from urllib import urlencode
from django.core.cache.backends.base import BaseCache
from django.utils.encoding import smart_str
#future-proofing against any Django upstream additions of new cache methods.
# If we didn't do this, we'd silently use the base class's methods, which wouldn't
# use the prefix as required.
_expected_methods = set(['__contains__', '__init__', 'add', 'close',
'decr', 'delete', 'get', 'get_many', 'has_key', 'incr', 'set',
'_get_memcache_timeout', 'clear', 'delete_many', 'set_many'])
class CacheClass(BaseCache):
def __init__(self, server, params):
#e.g. CACHE_BACKEND="path.to.prefixed://memcached://10.0.0.3:11211/?prefix=xxx"
# Falls back to settings.SITE_ID if no specific prefix is given.
from django.core.cache import get_cache
self.prefix = smart_str(params.pop('prefix', str(settings.SITE_ID) + ':'))
backend_uri = server
if params:
backend_uri += "?%s" % urlencode(params)
self.backend = get_cache(backend_uri)
actual_methods = set([m for m in dir(self.backend)
if isinstance(getattr(self.backend, m), types.MethodType)])
new_methods = actual_methods - _expected_methods
assert not bool(new_methods), "New methods have been added to cache backend: %s." % new_methods
super(CacheClass, self).__init__(params)
def get(self, key, default=None):
return self.backend.get(self.prefix + key, default)
def set(self, key, value, timeout=0):
return self.backend.set(self.prefix + key, value, timeout)
def delete(self, key):
self.backend.delete(self.prefix + key)
def add(self, key, value, timeout=None):
self.backend.add(self.prefix + key, value, timeout)
def get_many(self, keys):
l_prefix = len(self.prefix)
_keys = [(self.prefix + key) for key in keys]
_results = self.backend.get_many(_keys)
for k in _keys:
if k in _results:
_results[k[l_prefix:]] = _results[k]
del _results[k]
return _results
def has_key(self, key):
return self.backend.has_key(self.prefix + key)
def incr(self, key, delta=1):
return self.backend.incr(self.prefix + key, delta)
def decr(self, key, delta=1):
return self.backend.incr(self.prefix + key, -delta)
def __contains__(self, key):
return self.backend.has_key(self.prefix + key)
def delete_many(self, keys):
_keys = [(self.prefix + key) for key in keys]
self._cache.delete_multi(map(smart_str, _keys))
def set_many(self, data, timeout=0):
safe_data = {}
for key, value in data.items():
if isinstance(value, unicode):
value = value.encode('utf-8')
safe_data[smart_str(self.prefix + key)] = value
self._cache.set_multi(safe_data, self._get_memcache_timeout(timeout))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment