Merge pull request intercom#156 from jkeyes/scroll-api

Adding support for Scroll API.

Author: jkeyes

intercom/api_operations/scroll.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+"""Operation to scroll through users."""
+
+from intercom import utils
+from intercom.scroll_collection_proxy import ScrollCollectionProxy
+
+
+class Scroll(object):
+    """A mixin that provides `scroll` functionality."""
+
+    def scroll(self, **params):
+        """Find all instances of the resource based on the supplied parameters."""
+        collection_name = utils.resource_class_to_collection_name(
+            self.collection_class)
+        finder_url = "/{}/scroll".format(collection_name)
+        return ScrollCollectionProxy(
+            self.client, self.collection_class, collection_name, finder_url)

intercom/scroll_collection_proxy.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+"""Proxy for the Scroll API."""
+import six
+from intercom import HttpError
+
+
+class ScrollCollectionProxy(six.Iterator):
+    """A proxy to iterate over resources returned by the Scroll API."""
+
+    def __init__(self, client, resource_class, resource_name, scroll_url):
+        """Initialise the proxy."""
+        self.client = client
+
+        # resource name
+        self.resource_name = resource_name
+
+        # resource class
+        self.resource_class = resource_class
+
+        # the original URL to retrieve the resources
+        self.scroll_url = scroll_url
+
+        # the identity of the scroll, extracted from the response
+        self.scroll_param = None
+
+        # an iterator over the resources found in the response
+        self.resources = None
+
+        # a link to the next page of results
+        self.next_page = None
+
+    def __iter__(self):
+        """Return self as the proxy has __next__ implemented."""
+        return self
+
+    def __next__(self):
+        """Return the next resource from the response."""
+        if self.resources is None:
+            # get the first page of results
+            self.get_first_page()
+
+        # try to get a resource if there are no more in the
+        # current resource iterator (StopIteration is raised)
+        # try to get the next page of results first
+        try:
+            resource = six.next(self.resources)
+        except StopIteration:
+            self.get_next_page()
+            resource = six.next(self.resources)
+
+        instance = self.resource_class(**resource)
+        return instance
+
+    def __getitem__(self, index):
+        """Return an exact item from the proxy."""
+        for i in range(index):
+            six.next(self)
+        return six.next(self)
+
+    def get_first_page(self):
+        """Return the first page of results."""
+        return self.get_page(self.scroll_param)
+
+    def get_next_page(self):
+        """Return the next page of results."""
+        return self.get_page(self.scroll_param)
+
+    def get_page(self, scroll_param=None):
+        """Retrieve a page of results from the Scroll API."""
+        if scroll_param is None:
+            response = self.client.get(self.scroll_url, {})
+        else:
+            response = self.client.get(self.scroll_url, {'scroll_param': scroll_param})
+
+        if response is None:
+            raise HttpError('Http Error - No response entity returned')
+
+        # create the resource iterator
+        collection = response[self.resource_name]
+        self.resources = iter(collection)
+        # grab the next page URL if one exists
+        self.scroll_param = self.extract_scroll_param(response)
+
+    def records_present(self, response):
+        """Return whether there are resources in the response."""
+        return len(response.get(self.resource_name)) > 0
+
+    def extract_scroll_param(self, response):
+        """Extract the scroll_param from the response."""
+        if self.records_present(response):
+            return response.get('scroll_param')

intercom/service/user.py

@@ -8,11 +8,12 @@
 from intercom.api_operations.delete import Delete
 from intercom.api_operations.save import Save
 from intercom.api_operations.load import Load
+from intercom.api_operations.scroll import Scroll
 from intercom.extended_api_operations.tags import Tags
 from intercom.service.base_service import BaseService
 
 
-class User(BaseService, All, Find, FindAll, Delete, Save, Load, Submit, Tags):
+class User(BaseService, All, Find, FindAll, Delete, Save, Load, Submit, Tags, Scroll):
 
     @property
     def collection_class(self):

tests/unit/__init__.py

@@ -196,6 +196,24 @@ def page_of_users(include_next_link=False):
     return page
 
 
+def users_scroll(include_users=False):  # noqa
+    # a "page" of results from the Scroll API
+    if include_users:
+        users = [
+            get_user("[email protected]"),
+            get_user("[email protected]"),
+            get_user("[email protected]")
+        ]
+    else:
+        users = []
+
+    return {
+        "type": "user.list",
+        "scroll_param": "da6bbbac-25f6-4f07-866b-b911082d7",
+        "users": users
+    }
+
+
 def page_of_events(include_next_link=False):
     page = {
         "type": "event.list",

tests/unit/test_scroll_collection_proxy.py

@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+"""Test module for Scroll Collection Proxy."""
+import unittest
+
+from intercom import HttpError
+from intercom.client import Client
+from mock import call
+from mock import patch
+from nose.tools import assert_raises
+from nose.tools import eq_
+from nose.tools import istest
+from tests.unit import users_scroll
+
+
+class CollectionProxyTest(unittest.TestCase):  # noqa
+
+    def setUp(self):  # noqa
+        self.client = Client()
+
+    @istest
+    def it_stops_iterating_if_no_users_returned(self):  # noqa
+        body = users_scroll(include_users=False)
+        with patch.object(Client, 'get', return_value=body) as mock_method:
+            emails = [user.email for user in self.client.users.scroll()]
+            mock_method.assert_called('/users/scroll', {})
+            eq_(emails, [])  # noqa
+
+    @istest
+    def it_keeps_iterating_if_users_returned(self):  # noqa
+        page1 = users_scroll(include_users=True)
+        page2 = users_scroll(include_users=False)
+        side_effect = [page1, page2]
+        with patch.object(Client, 'get', side_effect=side_effect) as mock_method:  # noqa
+            emails = [user.email for user in self.client.users.scroll()]
+            eq_([call('/users/scroll', {}), call('/users/scroll', {'scroll_param': 'da6bbbac-25f6-4f07-866b-b911082d7'})],  # noqa
+                mock_method.mock_calls)
+            eq_(emails, ['[email protected]', '[email protected]', '[email protected]'])  # noqa
+
+    @istest
+    def it_supports_indexed_array_access(self):  # noqa
+        body = users_scroll(include_users=True)
+        with patch.object(Client, 'get', return_value=body) as mock_method:
+            eq_(self.client.users.scroll()[0].email, '[email protected]')
+            mock_method.assert_called_once_with('/users/scroll', {})
+            eq_(self.client.users.scroll()[1].email, '[email protected]')
+
+    @istest
+    def it_returns_one_page_scroll(self):  # noqa
+        body = users_scroll(include_users=True)
+        with patch.object(Client, 'get', return_value=body):
+            scroll = self.client.users.scroll()
+            scroll.get_next_page()
+            emails = [user['email'] for user in scroll.resources]
+            eq_(emails, ['[email protected]', '[email protected]', '[email protected]'])  # noqa
+
+    @istest
+    def it_keeps_iterating_if_called_with_scroll_param(self):  # noqa
+        page1 = users_scroll(include_users=True)
+        page2 = users_scroll(include_users=False)
+        side_effect = [page1, page2]
+        with patch.object(Client, 'get', side_effect=side_effect) as mock_method:  # noqa
+            scroll = self.client.users.scroll()
+            scroll.get_page()
+            scroll.get_page('da6bbbac-25f6-4f07-866b-b911082d7')
+            emails = [user['email'] for user in scroll.resources]
+            eq_(emails, [])  # noqa
+
+    @istest
+    def it_works_with_an_empty_list(self):  # noqa
+        body = users_scroll(include_users=False)
+        with patch.object(Client, 'get', return_value=body) as mock_method:  # noqa
+            scroll = self.client.users.scroll()
+            scroll.get_page()
+            emails = [user['email'] for user in scroll.resources]
+            eq_(emails, []) # noqa
+
+    @istest
+    def it_raises_an_http_error(self):  # noqa
+        with patch.object(Client, 'get', return_value=None) as mock_method:  # noqa
+            scroll = self.client.users.scroll()
+            with assert_raises(HttpError):
+                scroll.get_page()