Merge pull request #491 from kyleknap/pagination-model

Added a PaginatorModel

botocore/paginate.py

@@ -19,6 +19,19 @@
 from botocore.utils import set_value_from_jmespath, merge_dicts
 
 
+class PaginatorModel(object):
+    def __init__(self, paginator_config):
+        self._paginator_config = paginator_config['pagination']
+
+    def get_paginator(self, operation_name):
+        try:
+            single_paginator_config = self._paginator_config[operation_name]
+        except KeyError:
+            raise ValueError("Paginator for operation does not exist: %s"
+                             % operation_name)
+        return single_paginator_config
+
+
 class PageIterator(object):
     def __init__(self, method, input_token, output_token, more_results,
                  result_keys, non_aggregate_keys, limit_key, max_items,

botocore/session.py

@@ -35,6 +35,7 @@
 from botocore.parsers import ResponseParserFactory
 from botocore import regions
 from botocore.model import ServiceModel
+from botocore import paginate
 import botocore.service
 from botocore import waiter
 from botocore import retryhandler, translate
@@ -510,6 +511,14 @@ def get_waiter_model(self, service_name, api_version=None):
         waiter_config = loader.load_data(waiter_path)
         return waiter.WaiterModel(waiter_config)
 
+    def get_paginator_model(self, service_name, api_version=None):
+        loader = self.get_component('data_loader')
+        latest = loader.determine_latest('%s/%s' % (
+            self.provider.name, service_name), api_version)
+        paginator_path = latest.replace('.normal', '.paginators')
+        paginator_config = loader.load_data(paginator_path)
+        return paginate.PaginatorModel(paginator_config)
+
     def get_service_data(self, service_name, api_version=None):
         """
         Retrieve the fully merged data associated with a service.

tests/unit/test_paginate.py

@@ -14,11 +14,38 @@
 from tests import unittest
 from botocore.paginate import Paginator as FuturePaginator
 from botocore.paginate import DeprecatedPaginator as Paginator
+from botocore.paginate import PaginatorModel
 from botocore.exceptions import PaginationError
 from botocore.operation import Operation
 
 import mock
 
+
+class TestPaginatorModel(unittest.TestCase):
+    def setUp(self):
+        self.paginator_config = {}
+        self.paginator_config['pagination'] = {
+            'ListFoos': {
+                'output_token': 'NextToken',
+                'input_token': 'NextToken',
+                'result_key': 'Foo'
+            }
+        }
+        self.paginator_model = PaginatorModel(self.paginator_config)
+
+    def test_get_paginator(self):
+        paginator_config = self.paginator_model.get_paginator('ListFoos')
+        self.assertEqual(
+            paginator_config,
+            {'output_token': 'NextToken', 'input_token': 'NextToken',
+             'result_key': 'Foo'}
+        )
+
+    def test_get_paginator_no_exists(self):
+        with self.assertRaises(ValueError):
+            paginator_config = self.paginator_model.get_paginator('ListBars')
+
+
 # TODO: FuturePaginator tests should be merged into tests that used the renamed
 # Deprecated paginators when we completely remove the Deprecated
 # paginator class and make all of the tests use the actual Paginator class

tests/unit/test_session.py

@@ -27,6 +27,7 @@
 from botocore import client
 from botocore.hooks import HierarchicalEmitter
 from botocore.waiter import WaiterModel
+from botocore.paginate import PaginatorModel
 
 
 class BaseSessionTest(unittest.TestCase):
@@ -346,6 +347,22 @@ def test_get_service_model(self):
         self.assertEqual(model.service_name, 'made_up')
 
 
+class TestGetPaginatorModel(BaseSessionTest):
+    def test_get_paginator_model(self):
+        loader = mock.Mock()
+        loader.determine_latest.return_value = 'aws/foo/2014-01-01.normal.json'
+        loader.load_data.return_value = {"pagination": {}}
+        self.session.register_component('data_loader', loader)
+
+        model = self.session.get_paginator_model('foo')
+
+        # Verify we get a PaginatorModel back
+        self.assertIsInstance(model, PaginatorModel)
+        # Verify we called the loader correctly.
+        loader.load_data.assert_called_with(
+            'aws/foo/2014-01-01.paginators.json')
+
+
 class TestGetWaiterModel(BaseSessionTest):
     def test_get_waiter_model(self):
         loader = mock.Mock()