Memcached Property Attributes

memcached_descriptors.py

"""
    Property attributes in memcached

    :copyright: (c) 2011 by Ori Livneh
    :license: Public Domain
"""
from django.core.cache import cache


def memcached_property(key_function, timeout=cache.default_timeout, doc=None):
    """
    Creates a property attribute that is stored in memcached. `key_function`
    should be a callable that accepts an object instance and returns a string
    representing a memcached key. Example::

        class Book(models.Model):
            isbn = models.CharField(max_length=13, primary_key=True)
            last_read = memcached_property(
                lambda self: "book:%d:last_read" % self.isbn)

    :param key_function: a callable that accepts an object instance and returns
                         a string representing a memcached key.
    :param timeout: used by setter to specify expiry; defaults to Django
                    default.
    :param doc: If given, doc will be the docstring of the property attribute.
    """
    if doc is None:
        doc = memcached_property.__doc__

    def fget(self):
        """ Getter function """
        key = key_function(self)
        return cache.get(key)

    def fset(self, value):
        """ Setter function """
        key = key_function(self)
        cache.set(key, value, timeout)

    def fdel(self):
        """ Deleter function """
        key = key_function(self)
        cache.delete(key)

    return property(fget, fset, fdel, doc)


if __name__ == '__main__':
    # when invoked as a script, run unit test
    import unittest
    from datetime import datetime, timedelta

    class Book(object):
        """ Dummy class for testing memcached_property """
        def __init__(self, isbn):
            self.isbn = isbn
        last_read = memcached_property(lambda b: "book:%d:last_read" % b.isbn)

    class MemcachedPropertyTestCase(unittest.TestCase):
        """ Test case for memcached_property descriptor """
        def setUp(self):
            """ Delete leftover keys in memcache """
            cache.delete("book:12:last_read")
        tearDown = setUp

        def test_descriptor_access(self):
            """
            Tests getter, setter and deleter functions for memcached properties
            """
            book = Book(isbn=12)

            # test init
            self.assertIsNone(book.last_read)

            # test setter
            date = datetime.now()
            book.last_read = date
            self.assertEqual(book.last_read, date)
            self.assertEqual(cache.get("book:12:last_read"), date)

            # test that the getter function really does go to memcached by
            # changing the value in memcached directly.
            different_date = date - timedelta(hours=1)
            self.assertNotEqual(date, different_date)
            cache.set("book:12:last_read", different_date)
            self.assertEqual(book.last_read, different_date)

            # test the deleter function
            del book.last_read
            self.assertIsNone(book.last_read)
            # cache.add should return true if the key did not exist, thereby
            # conforming deletion.
            self.assertTrue(cache.add("book:12:last_read", 123))

    unittest.main()