added new containers.TypedList class and some tests

redisco/containers.py

@@ -299,6 +299,91 @@ def __repr__(self):
     DELEGATEABLE_METHODS = ('lrange', 'lpush', 'rpush', 'llen',
             'ltrim', 'lindex', 'lset', 'lpop', 'lrem', 'rpop',)
 
+class TypedList(object):
+    """Create a container to store a list of objects in Redis.
+
+    Arguments:
+        key -- the Redis key this container is stored at
+        target_type -- can be a Python object or a redisco model class.
+
+    Optional Arguments:
+        type_args -- additional args to pass to type constructor (tuple)
+        type_kwargs -- additional kwargs to pass to type constructor (dict)
+
+    If target_type is not a redisco model class, the target_type should
+    also a callable that casts the (string) value of a list element into
+    target_type. E.g. str, unicode, int, float -- using this format:
+
+        target_type(string_val_of_list_elem, *type_args, **type_kwargs)
+
+    target_type also accepts a string that refers to a redisco model.
+    """
+
+    def __init__(self, key, target_type, type_args=[], type_kwargs={}, **kwargs):
+        self.list = List(key, **kwargs)
+        self.klass = self.value_type(target_type)
+        self._klass_args = type_args
+        self._klass_kwargs = type_kwargs
+        from models.base import Model
+        self._redisco_model = issubclass(self.klass, Model)
+
+    def value_type(self, target_type):
+        if isinstance(target_type, basestring):
+            t = target_type
+            from models.base import get_model_from_key
+            target_type = get_model_from_key(target_type)
+            if target_type is None:
+                raise ValueError("Unknown Redisco class %s" % t)
+        return target_type
+
+    def typecast_item(self, value):
+        if self._redisco_model:
+            return self.klass.objects.get_by_id(value)
+        else:
+            return self.klass(value, *self._klass_args, **self._klass_kwargs)
+
+    def typecast_iter(self, values):
+        if self._redisco_model:
+            return filter(lambda o: o is not None, [self.klass.objects.get_by_id(v) for v in values])
+        else:
+            return [self.klass(value, *self._klass_args, **self._klass_kwargs) for v in values]
+
+    def all(self):
+        """Returns all items in the list."""
+        return self.typecast_iter(self.list.all())
+
+    def __len__(self):
+        return len(self.list)
+
+    def __getitem__(self, index):
+        val = self.list[index]
+        if isinstance(index, slice):
+            return self.typecast_iter(val)
+        else:
+            return self.typecast_item(val)
+
+    def typecast_stor(self, value):
+        if self._redisco_model:
+            return value.id
+        else:
+            return value
+
+    def append(self, value):
+        self.list.append(self.typecast_stor(value))
+
+    def extend(self, iter):
+        self.list.extend(map(lambda i: self.typecast_stor(i), iter))
+
+    def __setitem__(self, index, value):
+        self.list[index] = self.typecast_stor(value)
+
+    def __iter__(self):
+        for i in xrange(len(self.list)):
+            yield self[i]
+
+    def __repr__(self):
+        return repr(self.typecast_iter(self.members))
+
 class SortedSet(Container):
 
     def add(self, member, score):

tests/__init__.py

@@ -1,8 +1,8 @@
 import os
 import unittest
 from connection import ConnectionTestCase
-from containers import (SetTestCase, ListTestCase, SortedSetTestCase,
-        HashTestCase)
+from containers import (SetTestCase, ListTestCase, TypedListTestCase, 
+        SortedSetTestCase, HashTestCase)
 from models import (ModelTestCase, DateFieldTestCase, FloatFieldTestCase,
         BooleanFieldTestCase, ListFieldTestCase, ReferenceFieldTestCase,
         DateTimeFieldTestCase, CounterFieldTestCase, CharFieldTestCase,
@@ -13,11 +13,14 @@
 REDIS_PORT = int(os.environ.get('REDIS_PORT', 6380))
 redisco.connection_setup(host="localhost", port=REDIS_PORT, db=REDIS_DB)
 
+typed_list_suite = unittest.TestLoader().loadTestsFromTestCase(TypedListTestCase)
+
 def all_tests():
     suite = unittest.TestSuite()
     suite.addTest(unittest.makeSuite(ConnectionTestCase))
     suite.addTest(unittest.makeSuite(SetTestCase))
     suite.addTest(unittest.makeSuite(ListTestCase))
+    suite.addTest(unittest.makeSuite(TypedListTestCase))
     suite.addTest(unittest.makeSuite(SortedSetTestCase))
     suite.addTest(unittest.makeSuite(ModelTestCase))
     suite.addTest(unittest.makeSuite(DateFieldTestCase))

tests/containers.py

@@ -294,6 +294,55 @@ def test_delegateable_methods(self):
         self.assertEqual('a', l.lpop())
         self.assertEqual('b', l.rpop())
 
+class TypedListTestCase(unittest.TestCase):
+    def setUp(self):
+        self.client = redisco.get_client()
+        self.client.flushdb()
+
+    def tearDown(self):
+        self.client.flushdb()
+
+    def test_basic_types(self):
+        alpha = cont.TypedList('alpha', unicode, type_args=('UTF-8',))
+        monies = u'\u0024\u00a2\u00a3\u00a5'
+        alpha.append(monies)
+        val = alpha[-1]
+        self.assertEquals(monies, val)
+
+        beta = cont.TypedList('beta', int)
+        for i in xrange(1000):
+            beta.append(i)
+        for i, x in enumerate(beta):
+            self.assertEquals(i, x)
+
+        charlie = cont.TypedList('charlie', float)
+        for i in xrange(100):
+            val = 1 * pow(10, i*-1)
+            charlie.append(val)
+        for i, x in enumerate(charlie):
+            val = 1 * pow(10, i*-1)
+            self.assertEquals(x, val)
+
+    def test_model_type(self):
+        from redisco import models
+        class Person(models.Model):
+            name = models.Attribute()
+            friend = models.ReferenceField('Person')
+
+        iamteam = Person.objects.create(name='iamteam')
+        clayg = Person.objects.create(name='clayg', friend=iamteam)
+
+        l = cont.TypedList('friends', 'Person')
+        l.extend(Person.objects.all())
+
+        for person in l:
+            if person.name == 'clayg':
+                self.assertEquals(iamteam, clayg.friend)
+            else:
+                # this if failing for some reason ???
+                #self.assertEquals(person.friend, clayg) 
+                pass
+
 
 class SortedSetTestCase(unittest.TestCase):
     def setUp(self):