add a debug_object command - so we can get a key's size (among other things)

Author: Shaneal Manek

redis/client.py

@@ -44,6 +44,20 @@ def dict_merge(*dicts):
     [merged.update(d) for d in dicts]
     return merged
 
+def parse_debug_object(response):
+  "Parse the results of Redis's DEBUG OBJECT command into a Python dict"
+  res = dict([kv.split(':') for kv in ('type:' + response).split()])
+  
+  # parse some expected int values from the string response
+  # note: this cmd isn't spec'd so these may not appear in all redis versions
+  possible_int_fields = ['refcount', 'serializedlength', 
+                         'lru', 'lru_seconds_idle']
+  for field in possible_int_fields:
+    if field in res:
+      res[field] = int(res[field])
+
+  return res
+
 def parse_info(response):
     "Parse the result of Redis's INFO command into a Python dict"
     info = {}
@@ -151,6 +165,7 @@ class StrictRedis(object):
             'CONFIG': parse_config,
             'HGETALL': lambda r: r and pairs_to_dict(r) or {},
             'INFO': parse_info,
+            'DEBUG' : parse_debug_object,
             'LASTSAVE': timestamp_to_datetime,
             'PING': lambda r: r == 'PONG',
             'RANDOMKEY': lambda r: r and r or None,
@@ -309,6 +324,10 @@ def info(self):
         "Returns a dictionary containing information about the Redis server"
         return self.execute_command('INFO')
 
+    def debug_object(self, key):
+      """Returns version specific metainformation about a give key"""
+      return self.execute_command('DEBUG', 'OBJECT', key)
+
     def lastsave(self):
         """
         Return a Python datetime object representing the last time the

tests/server_commands.py

@@ -91,6 +91,15 @@ def test_info(self):
         self.assert_(isinstance(info, dict))
         self.assertEquals(info['db9']['keys'], 2)
 
+    def test_debug_object(self):
+      self.client['a'] = 'foo'
+      debug_info = self.client.debug_object('a')
+      self.assert_(len(debug_info) > 0)
+      self.assertEquals(1, debug_info['refcount'])
+      self.assert_(debug_info['serializedlength'] > 0)
+      self.client.rpush('b', 'a1')
+      debug_info = self.client.debug_object('a')
+      
     def test_lastsave(self):
         self.assert_(isinstance(self.client.lastsave(), datetime.datetime))