Adds 'ttl' method to return the number of seconds left on the key before it expires.

Author: sebleier

redis_cache/cache.py

@@ -321,6 +321,18 @@ def incr(self, key, delta=1, version=None):
             self.set(key, value)
         return value
 
+    def ttl(self, key, version=None):
+        """
+        Returns the 'time-to-live' of a key.  If the key is not volitile, i.e.
+        it has not set expiration, then the value returned is None.  Otherwise,
+        the value is the number of seconds remaining.  If the key does not exist,
+        0 is returned.
+        """
+        key = self.make_key(key, version=version)
+        if self._client.exists(key):
+            return self._client.ttl(key)
+        return 0
+
 
 class RedisCache(CacheClass):
     """
@@ -337,7 +349,7 @@ def incr_version(self, key, delta=1, version=None):
         Adds delta to the cache version for the supplied key. Returns the
         new version.
 
-        Note: In Redis 2.0 you cannot rename a volitle key, so we have to move
+        Note: In Redis 2.0 you cannot rename a volitile key, so we have to move
         the value from the old key to the new key and maintain the ttl.
         """
         if version is None:
@@ -353,3 +365,4 @@ def incr_version(self, key, delta=1, version=None):
         self.set(new_key, value, timeout=ttl)
         self.delete(old_key)
         return version + delta
+

tests/testapp/tests.py

@@ -391,6 +391,33 @@ def noop(*args, **kwargs):
         cache._client.connection_pool.release = release
         cache._client.connection_pool.max_connections = 2**31
 
+    def test_ttl_set_expiry(self):
+        self.cache.set('a', 'a', 10)
+        ttl = self.cache.ttl('a')
+        self.assertAlmostEqual(ttl, 10)
+
+    def test_ttl_no_expiry(self):
+        self.cache.set('a', 'a', timeout=None)
+        ttl = self.cache.ttl('a')
+        self.assertTrue(ttl is None)
+
+    def test_ttl_past_expiry(self):
+        self.cache.set('a', 'a', timeout=1)
+        ttl = self.cache.ttl('a')
+        self.assertAlmostEqual(ttl, 1)
+
+        time.sleep(1)
+
+        ttl = self.cache.ttl('a')
+        self.assertEqual(ttl, 0)
+
+    def test_non_existent_key(self):
+        """ Non-existent keys are semantically the same as keys that have
+        expired.
+        """
+        ttl = self.cache.ttl('does_not_exist')
+        self.assertEqual(ttl, 0)
+
 
 if __name__ == '__main__':
     import unittest