8
\$\begingroup\$

For a session implementation I needed a property that caches its getter (since it involves a database lookup) but still allows modifications (e.g. assigning a new user and storing that user's id in the session). To make this as comfortable as possibly I subclassed property and added the caching logic to the descriptor methods.

While it works fine and I think I covered all important cases with my unittest I'd prefer if someone else looked over it, too, just in case I missed something.

The code

def cached_writable_property(cache_attr, cache_on_set=True):
    class _cached_writable_property(property):
        def __get__(self, obj, objtype=None):
            if obj is not None and self.fget and hasattr(obj, cache_attr):
                return getattr(obj, cache_attr)
            value = property.__get__(self, obj, objtype)
            setattr(obj, cache_attr, value)
            return value

        def __set__(self, obj, value):
            property.__set__(self, obj, value)
            if cache_on_set:
                setattr(obj, cache_attr, value)
            else:
                try:
                    delattr(obj, cache_attr)
                except AttributeError:
                    pass

        def __delete__(self, obj):
            property.__delete__(self, obj)
            try:
                delattr(obj, cache_attr)
            except AttributeError:
                pass

    return _cached_writable_property

The unittest

if __name__ == '__main__':
    import unittest

    class CannotSet(Exception):
        pass

    class _Test(dict):
        def __init__(self, *args, **kw):
            dict.__init__(self, *args, **kw)
            self.log = []

        @cached_writable_property('_a', True)
        def a(self):
            self.log.append('get a')
            return self['a'][0]
        @a.setter
        def a(self, val):
            self.log.append('set a')
            self['a'] = (val, 'xyz')
        @a.deleter
        def a(self):
            self.log.append('del a')
            del self['a']


        @cached_writable_property('_b', False)
        def b(self):
            self.log.append('get b')
            return self['b'][0]
        @b.setter
        def b(self, val):
            self.log.append('set b')
            self['b'] = (val, 'lol')

        @cached_writable_property('_c')
        def c(self):
            self.log.append('get c')
            return self['c'][0]
        @c.setter
        def c(self, val):
            self.log.append('set c')
            raise CannotSet


    class TestCachedWritableProperty(unittest.TestCase):
        def test_cwp(self):
            _log = []
            def _expect_log(obj, entry):
                if entry is not None:
                    _log.append(entry)
                self.assertEqual(obj.log, _log)

            t = _Test(b=('initial-b',))
            # Does not exist => error from getter
            self.assertRaises(KeyError, lambda: t.a)
            _expect_log(t, 'get a')
            # Exists but not cached => result from getter
            self.assertEquals(t.b, 'initial-b')
            _expect_log(t, 'get b')
            # Exists and cached => result from cache
            self.assertEquals(t.b, 'initial-b')
            _expect_log(t, None)

            # Modify, cache on set
            t.a = 'new-a'
            _expect_log(t, 'set a')
            # Read, from cache
            self.assertEquals(t.a, 'new-a')
            _expect_log(t, None)
            # Read, from cache again
            self.assertEquals(t.a, 'new-a')
            _expect_log(t, None)

            # Delete, should be removed from cache, too
            del t.a
            _expect_log(t, 'del a')
            self.assertRaises(KeyError, lambda: t.a)
            self.assertRaises(KeyError, lambda: t['a'])
            _expect_log(t, 'get a')

            # Modify, do not cache on set
            t.b = 'new-b'
            _expect_log(t, 'set b')
            # Read, not from cache
            self.assertEquals(t.b, 'new-b')
            _expect_log(t, 'get b')
            # Read, from cache
            self.assertEquals(t.b, 'new-b')
            _expect_log(t, None)

            # Test failing setters
            self.assertRaises(KeyError, lambda: t.c)
            _expect_log(t, 'get c')
            self.assertRaises(CannotSet, lambda: setattr(t, 'c', 'nope'))
            _expect_log(t, 'set c')
            t['c'] = ('initial-c',)
            self.assertEquals(t.c, 'initial-c')
            _expect_log(t, 'get c')
            self.assertRaises(CannotSet, lambda: setattr(t, 'c', 'nope'))
            _expect_log(t, 'set c')
            self.assertEquals(t.c, 'initial-c')
            _expect_log(t, None)


    unittest.main()
\$\endgroup\$
3
  • \$\begingroup\$ Not sure if this'll be used in a multi-threaded context, but if so you may want to a) lock around the cache accesses and b) invalidate the cache before setting/deleting, instead of before. \$\endgroup\$ – disatisfieddinosaur Aug 12 '13 at 5:32
  • 1
    \$\begingroup\$ Django has an implementation of a cached property here: docs.djangoproject.com/en/dev/ref/utils/…. You can set it and clear it by deleting the attribute on the object. I don't think it has an dependencies in Django, so you may be able to just take the code and use it in any python project. \$\endgroup\$ – hgcrpd Sep 17 '13 at 13:25
  • \$\begingroup\$ Django's cached_property modifies the instance to replace the property with the cached value (see the source). So it can only ever be used once, whereas the cached property in the OP can be deleted, which will cause it to be re-computed on the next lookup. \$\endgroup\$ – Gareth Rees Dec 17 '13 at 12:30
2
\$\begingroup\$

If you can afford to make your class hashable, and you read more often than you need to update the values inside (because an update to any property will invalidate the cache for all the other ones, and for this reason I set the maxsize to 1), it's probably much simpler to just combine property and functools.lru_cache

from functools import lru_cache

if __name__ == '__main__':
    import unittest

    class CannotSet(Exception):
        pass

    class _Test(dict):
        def __init__(self, *args, **kw):
            dict.__init__(self, *args, **kw)
            self.log = []

        def __hash__(self):
            return hash(tuple(sorted(self.items())))

        @property
        @lru_cache(1)
        def a(self):
            self.log.append('get a')
            return self['a'][0]
        @a.setter
        def a(self, val):
            self.log.append('set a')
            self['a'] = (val, 'xyz')
        @a.deleter
        def a(self):
            self.log.append('del a')
            del self['a']


        @property
        @lru_cache(1)
        def b(self):
            self.log.append('get b')
            return self['b'][0]
        @b.setter
        def b(self, val):
            self.log.append('set b')
            self['b'] = (val, 'lol')

        @property
        @lru_cache(1)
        def c(self):
            self.log.append('get c')
            return self['c'][0]
        @c.setter
        def c(self, val):
            self.log.append('set c')
            raise CannotSet


    class TestCachedWritableProperty(unittest.TestCase):
        def test_cwp(self):
            _log = []
            def _expect_log(obj, entry):
                if entry is not None:
                    _log.append(entry)
                self.assertEqual(obj.log, _log)

            t = _Test(b=('initial-b',))
            # Does not exist => error from getter
            self.assertRaises(KeyError, lambda: t.a)
            _expect_log(t, 'get a')
            # Exists but not cached => result from getter
            self.assertEquals(t.b, 'initial-b')
            _expect_log(t, 'get b')
            # Exists and cached => result from cache
            self.assertEquals(t.b, 'initial-b')
            _expect_log(t, None)

            # Modify, cache on set
            t.a = 'new-a'
            _expect_log(t, 'set a')
            # Read
            self.assertEquals(t.a, 'new-a')
            _expect_log(t, 'get a')
            # Read, from cache again
            self.assertEquals(t.a, 'new-a')
            _expect_log(t, None)

            # Delete, should be removed from cache, too
            del t.a
            _expect_log(t, 'del a')
            self.assertRaises(KeyError, lambda: t.a)
            self.assertRaises(KeyError, lambda: t['a'])
            _expect_log(t, 'get a')

            # Modify, do not cache on set
            t.b = 'new-b'
            _expect_log(t, 'set b')
            # Read, not from cache
            self.assertEquals(t.b, 'new-b')
            _expect_log(t, 'get b')
            # Read, from cache
            self.assertEquals(t.b, 'new-b')
            _expect_log(t, None)

            # Test failing setters
            self.assertRaises(KeyError, lambda: t.c)
            _expect_log(t, 'get c')
            self.assertRaises(CannotSet, lambda: setattr(t, 'c', 'nope'))
            _expect_log(t, 'set c')
            t['c'] = ('initial-c',)
            self.assertEquals(t.c, 'initial-c')
            _expect_log(t, 'get c')
            self.assertRaises(CannotSet, lambda: setattr(t, 'c', 'nope'))
            _expect_log(t, 'set c')
            self.assertEquals(t.c, 'initial-c')
            _expect_log(t, None)


    unittest.main()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.