- add WeakSequence.append()

- fix and test weakref cleanout for WeakSequence, [ticket:2794]

diff --git a/lib/sqlalchemy/util/_collections.py b/lib/sqlalchemy/util/_collections.py
index b2f03450decee556ab1b5d581f493b2a038e15bf..90d9f91f90338bbdbd04e67e1c438e650ca1af71 100644 (file)
--- a/lib/sqlalchemy/util/_collections.py
+++ b/lib/sqlalchemy/util/_collections.py
@@ -617,16 +617,23 @@ class IdentitySet(object):
 
 
 class WeakSequence(object):
-    def __init__(self, elements):
+    def __init__(self, __elements=()):
         self._storage = [
-            weakref.ref(element) for element in elements
+            weakref.ref(element, self._remove) for element in __elements
         ]
 
+    def append(self, item):
+        self._storage.append(weakref.ref(item, self._remove))
+
     def _remove(self, ref):
         self._storage.remove(ref)
 
+    def __len__(self):
+        return len(self._storage)
+
     def __iter__(self):
-        return (obj for obj in (ref() for ref in self._storage) if obj is not None)
+        return (obj for obj in
+                    (ref() for ref in self._storage) if obj is not None)
 
     def __getitem__(self, index):
         try:

diff --git a/test/base/test_utils.py b/test/base/test_utils.py
index b28d26e712637c99b86fe2a74467872ce83cd72a..0b7fa7072a7aa88491f41a02845cb47c394b24ef 100644 (file)
--- a/test/base/test_utils.py
+++ b/test/base/test_utils.py
@@ -1,10 +1,10 @@
 import copy
 
-from sqlalchemy import util, sql, exc
+from sqlalchemy import util, sql, exc, testing
 from sqlalchemy.testing import assert_raises, assert_raises_message, fixtures
 from sqlalchemy.testing import eq_, is_, ne_, fails_if
-from sqlalchemy.testing.util import picklers
-from sqlalchemy.util import classproperty
+from sqlalchemy.testing.util import picklers, gc_collect
+from sqlalchemy.util import classproperty, WeakSequence
 
 
 class KeyedTupleTest():
@@ -115,6 +115,36 @@ class KeyedTupleTest():
             keyed_tuple[0] = 100
         assert_raises(TypeError, should_raise)
 
+class WeakSequenceTest(fixtures.TestBase):
+    @testing.requires.predictable_gc
+    def test_cleanout_elements(self):
+        class Foo(object):
+            pass
+        f1, f2, f3 = Foo(), Foo(), Foo()
+        w = WeakSequence([f1, f2, f3])
+        eq_(len(w), 3)
+        eq_(len(w._storage), 3)
+        del f2
+        gc_collect()
+        eq_(len(w), 2)
+        eq_(len(w._storage), 2)
+
+    @testing.requires.predictable_gc
+    def test_cleanout_appended(self):
+        class Foo(object):
+            pass
+        f1, f2, f3 = Foo(), Foo(), Foo()
+        w = WeakSequence()
+        w.append(f1)
+        w.append(f2)
+        w.append(f3)
+        eq_(len(w), 3)
+        eq_(len(w._storage), 3)
+        del f2
+        gc_collect()
+        eq_(len(w), 2)
+        eq_(len(w._storage), 2)
+
 
 class OrderedDictTest(fixtures.TestBase):