ListAllTables!

LowVoltage/__init__.py

@@ -7,3 +7,4 @@
 from actions import *
 from exceptions import *
 from attribute_types import *
+from iterators import *

LowVoltage/iterators/__init__.py

@@ -0,0 +1,5 @@
+# coding: utf8
+
+# Copyright 2014-2015 Vincent Jacques <vincent@vincent-jacques.net>
+
+from .list_all_tables import ListAllTables

LowVoltage/iterators/list_all_tables.py

@@ -0,0 +1,72 @@
+# coding: utf8
+
+# Copyright 2014-2015 Vincent Jacques <vincent@vincent-jacques.net>
+
+import unittest
+
+import LowVoltage as _lv
+import LowVoltage.testing as _tst
+
+
+class ListAllTables(object):
+    """Make as many ListTables as needed to iterate over all tables"""
+    # Don't implement anything else than forward iteration. (Remember PyGithub's PaginatedList; it was too difficult to maintain for niche use-cases)
+    # Clients can use raw ListTables actions to implement their specific needs.
+
+    def __init__(self, connection):
+        self.__connection = connection
+        self.__current_iter = [].__iter__()
+        self.__next_start_table = None
+        self.__done = False
+
+    def __iter__(self):
+        return self
+
+    def next(self):
+        try:
+            return self.__current_iter.next()
+        except StopIteration:
+            if self.__done:
+                raise
+            else:
+                r = self.__connection.request(_lv.ListTables().exclusive_start_table_name(self.__next_start_table))
+                self.__next_start_table = r.last_evaluated_table_name
+                self.__done = self.__next_start_table is None
+                self.__current_iter = r.table_names.__iter__()
+                return self.__current_iter.next()
+
+
+class ListAllTablesLocalIntegTests(_tst.dynamodb_local.TestCase):
+    def __test(self, tables_count):
+        table_names = ["Tab{:03}".format(i) for i in range(tables_count)]
+
+        for t in table_names:
+            self.connection.request(
+                _lv.CreateTable(t).hash_key("h", _lv.STRING).provisioned_throughput(1, 1)
+            )
+
+        self.assertEqual(
+            list(_lv.ListAllTables(self.connection)),
+            table_names
+        )
+
+        for t in table_names:
+            self.connection.request(_lv.DeleteTable(t))
+
+    def test_list_0_tables(self):
+        self.__test(0)
+
+    def test_list_99_tables(self):
+        self.__test(99)
+
+    def test_list_100_tables(self):
+        self.__test(100)
+
+    def test_list_101_tables(self):
+        self.__test(101)
+
+    def test_list_102_tables(self):
+        self.__test(102)
+
+    def test_list_250_tables(self):
+        self.__test(250)

LowVoltage/iterators/tests/__init__.py

@@ -0,0 +1,3 @@
+# coding: utf8
+
+# Copyright 2014-2015 Vincent Jacques <vincent@vincent-jacques.net>

LowVoltage/iterators/tests/integ/__init__.py

@@ -0,0 +1,3 @@
+# coding: utf8
+
+# Copyright 2014-2015 Vincent Jacques <vincent@vincent-jacques.net>

LowVoltage/iterators/tests/integ/local.py

@@ -0,0 +1,5 @@
+# coding: utf8
+
+# Copyright 2014-2015 Vincent Jacques <vincent@vincent-jacques.net>
+
+from ...list_all_tables import ListAllTablesLocalIntegTests

LowVoltage/iterators/tests/unit.py

@@ -0,0 +1,3 @@
+# coding: utf8
+
+# Copyright 2014-2015 Vincent Jacques <vincent@vincent-jacques.net>

LowVoltage/tests/integ/local.py

@@ -7,6 +7,7 @@
 import LowVoltage.testing.dynamodb_local
 from LowVoltage.actions.tests.integ.local import *
 from LowVoltage.connection.tests.integ.local import *
+from LowVoltage.iterators.tests.integ.local import *
 
 
 if __name__ == "__main__":  # pragma no branch (Test code)

LowVoltage/tests/unit.py

@@ -8,6 +8,7 @@
 from LowVoltage.testing.tests.unit import *
 from LowVoltage.policies.tests.unit import *
 from LowVoltage.connection.tests.unit import *
+from LowVoltage.iterators.tests.unit import *
 
 
 if __name__ == "__main__":  # pragma no branch (Test code)