From 30463ab5ab93b2c906721b9b0b217d9a6e944d12 Mon Sep 17 00:00:00 2001
From: "Daniel G. Taylor" <danielgtaylor@gmail.com>
Date: Wed, 26 Nov 2014 15:22:12 -0800
Subject: [PATCH] Implement batch actions.

This change implements batch actions on collections:

```python
s3.Bucket('boto3').objects.delete()
```

It does the following:

* Create a `CollectionFactory` class to generate subclasses of
  `CollectionManager` and `ResourceCollection`.
* Update the `ResourceFactory` to use the new `CollectionFactory`.
* Add a public `pages()` method to collections. This returns entire
  pages of resource instances and it used by `__iter__` as well.
* Add batch actions as methods via the new `CollectionFactory`.
* Add a `BatchAction` subclass of `Action` which does the following:

    1. Get a page of results from the collection's operation
    2. Build parameters for the batch action operation
    3. Call the batch action operation
    4. Repeat until no more pages

* Makes some previously public members private on `Action` as these
  should never have been public.
* Update documentation to include collection classes.
* Add tests to cover the new functionality.
---
 boto3/docs.py                           |  90 ++++++++++-
 boto3/resources/action.py               |  84 +++++++++-
 boto3/resources/collection.py           | 204 ++++++++++++++++++++----
 boto3/resources/factory.py              |  19 ++-
 boto3/resources/model.py                |  11 +-
 boto3/resources/params.py               |  12 +-
 docs/source/guide/collections.rst       |   4 +
 tests/unit/resources/test_action.py     | 116 +++++++++++++-
 tests/unit/resources/test_collection.py | 159 +++++++++++++++++-
 tests/unit/resources/test_factory.py    |   9 +-
 tests/unit/resources/test_params.py     |  34 +++-
 11 files changed, 685 insertions(+), 57 deletions(-)

diff --git a/boto3/docs.py b/boto3/docs.py
index ee83aafbdf..55c7004460 100644
--- a/boto3/docs.py
+++ b/boto3/docs.py
@@ -131,7 +131,18 @@ def docs_for(service_name):
 
     print('Processing {0}-{1}'.format(service_name, service_model.api_version))
 
+    # The following creates an official name of 'Amazon Simple Storage
+    # Service (S3)' our of 'Amazon Simple Storage Service' and 'Amazon S3'.
+    # It tries to be smart, so for `Amazon DynamoDB' and 'DynamoDB' we would
+    # get an official name of 'Amazon DynamoDB'.
     official_name = service_model.metadata.get('serviceFullName')
+    short_name = service_model.metadata.get('serviceAbbreviation', '')
+    if short_name.startswith('Amazon'):
+        short_name = short_name[7:]
+    if short_name.startswith('AWS'):
+        short_name = short_name[4:]
+    if short_name and short_name.lower() not in official_name.lower():
+        official_name += ' ({0})'.format(short_name)
 
     docs = '{0}\n{1}\n\n'.format(official_name, '=' * len(official_name))
 
@@ -142,18 +153,47 @@ def docs_for(service_name):
     filename = (os.path.dirname(__file__) + '/data/resources/'
                 '{0}-{1}.resources.json').format(service_name,
                                                  service_model.api_version)
+    # We can't use a set here because dicts aren't hashable!
+    models = {}
     if os.path.exists(filename):
         data = json.load(open(filename))
         model = ResourceModel(service_name, data['service'], data['resources'])
 
+        for collection_model in model.collections:
+            collection_model.parent_name = model.name
+            models[collection_model.name] = {
+                'type': 'collection',
+                'model': collection_model
+            }
+
         docs += document_resource(service_name, official_name, model,
                                   service_model)
 
+        # First, collect all the models...
         for name, model in sorted(data['resources'].items(),
                                   key=lambda i:i[0]):
             resource_model = ResourceModel(name, model, data['resources'])
-            docs += document_resource(service_name, official_name,
-                                      resource_model, service_model)
+            if name not in models:
+                models[name] = {'type': 'resource', 'model': resource_model}
+
+            for collection_model in resource_model.collections:
+                collection_model.parent_name = xform_name(resource_model.name)
+
+                cname = collection_model.name + 'CollectionManager'
+                if cname not in models:
+                    models[cname] = {'type': 'collection',
+                                     'model': collection_model}
+
+        # Then render them out in alphabetical order.
+        for name, item in sorted(models.items()):
+            model = item['model']
+            if item['type'] == 'resource':
+                docs += document_resource(service_name, official_name,
+                                          model, service_model)
+            elif item['type'] == 'collection':
+                docs += document_collection(
+                    service_name, official_name, model,
+                    model.resource.model, service_model)
 
     return docs
 
@@ -320,16 +360,53 @@ def document_resource(service_name, official_name, resource_model,
         for collection in sorted(resource_model.collections,
                                  key=lambda i: i.name):
             docs += ('   .. py:attribute:: {0}\n\n      '
-                     '(:py:class:`~boto3.resources.collection.CollectionManager`)'
-                     ' A collection of :py:class:`{1}.{2}` instances. This'
-                     ' collection uses the :py:meth:`{3}.Client.{4}` operation'
+                     '(:py:class:`{1}.{2}CollectionManager`)'
+                     ' A collection of :py:class:`{3}.{4}` instances. This'
+                     ' collection uses the :py:meth:`{5}.Client.{6}` operation'
                      ' to get items.\n\n').format(
                         xform_name(collection.name), service_name,
+                        collection.name, service_name,
                         collection.resource.type, service_name,
                         xform_name(collection.request.operation))
 
     return docs
 
+def document_collection(service_name, official_name, collection_model,
+                        resource_model, service_model):
+    """
+    Generate reference documentation about a collection and any
+    batch actions it might have.
+    """
+    title = collection_model.name + 'Collection'
+    docs = '{0}\n{1}\n\n'.format(title, '-' * len(title))
+    docs += '.. py:class:: {0}.{1}CollectionManager()\n\n'.format(
+        service_name, collection_model.name)
+    docs += ('   A collection of :py:class:`{0}.{1}` resources for {2}. See'
+             '   the'
+             '   :py:class:`~boto3.resources.collection.CollectionManager`'
+             '   base class for additional methods.\n\n'
+             '   This collection uses the :py:meth:`{3}.Client.{4}`'
+             '   operation to get items, and its parameters can be'
+             '   used as filters::\n\n').format(
+                service_name, resource_model.name, official_name,
+                service_name, xform_name(collection_model.request.operation))
+    docs += ('       for {0} in {1}.{2}.all():\n'
+             '           print({0})\n\n').format(
+                xform_name(collection_model.resource.type),
+                collection_model.parent_name,
+                xform_name(collection_model.name),
+                xform_name(collection_model.resource.type))
+
+    if collection_model.batch_actions:
+        docs += ('   .. rst-class:: admonition-title\n\n   Batch Actions\n\n'
+                 '   Batch actions provide a way to manipulate groups of'
+                 '   resources in a single service operation call.\n\n')
+    for action in sorted(collection_model.batch_actions, key=lambda i:i.name):
+        docs += document_action(action, service_name, resource_model,
+                                service_model)
+
+    return docs
+
 def document_action(action, service_name, resource_model, service_model,
                     action_type='action'):
     """
@@ -343,7 +420,8 @@ def document_action(action, service_name, resource_model, service_model,
         print('Cannot get operation ' + action.request.operation)
         return ''
 
-    ignore_params = [p.target for p in action.request.params]
+    # Here we split because we only care about top-level parameter names
+    ignore_params = [p.target.split('.')[0] for p in action.request.params]
 
     rtype = 'dict'
     if action_type == 'action':
diff --git a/boto3/resources/action.py b/boto3/resources/action.py
index 9a096f2ef6..b491f8322b 100644
--- a/boto3/resources/action.py
+++ b/boto3/resources/action.py
@@ -41,34 +41,34 @@ class ServiceAction(object):
     """
     def __init__(self, action_model, factory=None, resource_defs=None,
                  service_model=None):
-        self.action_model = action_model
+        self._action_model = action_model
 
         # In the simplest case we just return the response, but if a
         # resource is defined, then we must create these before returning.
         resource = action_model.resource
         if resource:
-            self.response_handler = ResourceHandler(resource.path,
+            self._response_handler = ResourceHandler(resource.path,
                 factory, resource_defs, service_model, resource,
                 action_model.request.operation)
         else:
-            self.response_handler = RawHandler(action_model.path)
+            self._response_handler = RawHandler(action_model.path)
 
     def __call__(self, parent, *args, **kwargs):
         """
         Perform the action's request operation after building operation
         parameters and build any defined resources from the response.
 
-        :type parent: ServiceResource
+        :type parent: :py:class:`~boto3.resources.base.ServiceResource`
         :param parent: The resource instance to which this action is attached.
         :rtype: dict or ServiceResource or list(ServiceResource)
         :return: The response, either as a raw dict or resource instance(s).
         """
-        operation_name = xform_name(self.action_model.request.operation)
+        operation_name = xform_name(self._action_model.request.operation)
 
         # First, build predefined params and then update with the
         # user-supplied kwargs, which allows overriding the pre-built
         # params if needed.
-        params = create_request_parameters(parent, self.action_model.request)
+        params = create_request_parameters(parent, self._action_model.request)
         params.update(kwargs)
 
         logger.info('Calling %s:%s with %r', parent.meta['service_name'],
@@ -78,4 +78,74 @@ def __call__(self, parent, *args, **kwargs):
 
         logger.debug('Response: %r', response)
 
-        return self.response_handler(parent, params, response)
+        return self._response_handler(parent, params, response)
+
+
+class BatchAction(ServiceAction):
+    """
+    An action which operates on a batch of items in a collection, typically
+    a single page of results from the collection's underlying service
+    operation call. For example, this allows you to delete up to 999
+    S3 objects in a single operation rather than calling ``.delete()`` on
+    each one individually.
+
+    :type action_model: :py:class:`~boto3.resources.model.Action`
+    :param action_model: The action model.
+    :type factory: ResourceFactory
+    :param factory: The factory that created the resource class to which
+                    this action is attached.
+    :type resource_defs: dict
+    :param resource_defs: Service resource definitions.
+    :type service_model: :ref:`botocore.model.ServiceModel`
+    :param service_model: The Botocore service model
+    """
+    def __call__(self, parent, *args, **kwargs):
+        """
+        Perform the batch action's operation on every page of results
+        from the collection.
+
+        :type parent: :py:class:`~boto3.resources.collection.ResourceCollection`
+        :param parent: The collection iterator to which this action
+                       is attached.
+        :rtype: list(dict)
+        :return: A list of low-level response dicts from each call.
+        """
+        service_name = None
+        client = None
+        responses = []
+        operation_name = xform_name(self._action_model.request.operation)
+
+        # Unlike the simple action above, a batch action must operate
+        # on batches (or pages) of items. So we get each page, construct
+        # the necessary parameters and call the batch operation.
+        for page in parent.pages():
+            params = {}
+            for resource in page:
+                # There is no public interface to get a service name
+                # or low-level client from a collection, so we get
+                # these from the first resource in the collection.
+                if service_name is None:
+                    service_name = resource.meta['service_name']
+                if client is None:
+                    client = resource.meta['client']
+
+                create_request_parameters(
+                    resource, self._action_model.request, params=params)
+
+            if not params:
+                # There are no items, no need to make a call.
+                break
+
+            params.update(kwargs)
+
+            logger.info('Calling %s:%s with %r',
+                        service_name, operation_name, params)
+
+            response = getattr(client, operation_name)(**params)
+
+            logger.debug('Response: %r', response)
+
+            responses.append(
+                self._response_handler(parent, params, response))
+
+        return responses
diff --git a/boto3/resources/collection.py b/boto3/resources/collection.py
index 52c0613b67..bbbb406729 100644
--- a/boto3/resources/collection.py
+++ b/boto3/resources/collection.py
@@ -16,6 +16,7 @@
 
 from botocore import xform_name
 
+from .action import BatchAction
 from .params import create_request_parameters
 from .response import ResourceHandler
 
@@ -63,6 +64,72 @@ def __iter__(self):
         A generator which yields resource instances after doing the
         appropriate service operation calls and handling any pagination
         on your behalf.
+
+        Page size, item limit, and filter parameters are applied
+        if they have previously been set.
+
+            >>> bucket = s3.Bucket('boto3')
+            >>> for obj in bucket.objects.all():
+            ...     print(obj.key)
+            'key1'
+            'key2'
+
+        """
+        limit = self._params.get('limit', None)
+
+        count = 0
+        for page in self.pages():
+            for item in page:
+                yield item
+
+                # If the limit is set and has been reached, then
+                # we stop processing items here.
+                count += 1
+                if limit is not None and count >= limit:
+                    return
+
+    def _clone(self, **kwargs):
+        """
+        Create a clone of this collection. This is used by the methods
+        below to provide a chainable interface that returns copies
+        rather than the original. This allows things like:
+
+            >>> base = collection.filter(Param1=1)
+            >>> query1 = base.filter(Param2=2)
+            >>> query2 = base.filter(Param3=3)
+            >>> query1.params
+            {'Param1': 1, 'Param2': 2}
+            >>> query2.params
+            {'Param1': 1, 'Param3': 3}
+
+        :rtype: :py:class:`ResourceCollection`
+        :return: A clone of this resource collection
+        """
+        params = copy.deepcopy(self._params)
+        params.update(kwargs)
+        clone = self.__class__(self._model, self._parent,
+                               self._handler, **params)
+        return clone
+
+    def pages(self):
+        """
+        A generator which yields pages of resource instances after
+        doing the appropriate service operation calls and handling
+        any pagination on your behalf. Non-paginated calls will
+        return a single page of items.
+
+        Page size, item limit, and filter parameters are applied
+        if they have previously been set.
+
+            >>> bucket = s3.Bucket('boto3')
+            >>> for page in bucket.objects.pages():
+            ...     for obj in page:
+            ...         print(obj.key)
+            'key1'
+            'key2'
+
+        :rtype: list(:py:class:`~boto3.resources.base.ServiceResource`)
+        :return: List of resource instances
         """
         client = self._parent.meta['client']
         cleaned_params = self._params.copy()
@@ -95,37 +162,21 @@ def __iter__(self):
         # we start processing and yielding individual items.
         count = 0
         for page in pages:
+            page_items = []
             for item in self._handler(self._parent, params, page):
-                yield item
+                page_items.append(item)
 
                 # If the limit is set and has been reached, then
                 # we stop processing items here.
                 count += 1
                 if limit is not None and count >= limit:
-                    return
+                    break
 
-    def _clone(self, **kwargs):
-        """
-        Create a clone of this collection. This is used by the methods
-        below to provide a chainable interface that returns copies
-        rather than the original. This allows things like:
+            yield page_items
 
-            >>> base = collection.filter(Param1=1)
-            >>> query1 = base.filter(Param2=2)
-            >>> query2 = base.filter(Param3=3)
-            >>> query1.params
-            {'Param1': 1, 'Param2': 2}
-            >>> query2.params
-            {'Param1': 1, 'Param3': 3}
-
-        :rtype: :py:class:`ResourceCollection`
-        :return: A clone of this resource collection
-        """
-        params = copy.deepcopy(self._params)
-        params.update(kwargs)
-        clone = self.__class__(self._model, self._parent,
-                               self._handler, **params)
-        return clone
+            # Stop reading pages if we've reached out limit
+            if limit is not None and count >= limit:
+                break
 
     def all(self, limit=None, page_size=None):
         """
@@ -182,6 +233,14 @@ def limit(self, count):
         """
         Return at most this many resources.
 
+            >>> for bucket in s3.buckets.limit(5):
+            ...     print(bucket.name)
+            'bucket1'
+            'bucket2'
+            'bucket3'
+            'bucket4'
+            'bucket5'
+
         :type count: int
         :param count: Return no more than this many items
         :rtype: :py:class:`ResourceCollection`
@@ -192,6 +251,9 @@ def page_size(self, count):
         """
         Fetch at most this many resources per service request.
 
+            >>> for obj in s3.Bucket('boto3').objects.page_size(100):
+            ...     print(obj.key)
+
         :type count: int
         :param count: Fetch this many items per request
         :rtype: :py:class:`ResourceCollection`
@@ -217,12 +279,18 @@ class CollectionManager(object):
         >>> for queue in sqs.queues.filter(QueueNamePrefix='AWS'):
         ...     print(queue.url)
 
+    Get whole pages of items:
+
+        >>> for page in s3.Bucket('boto3').objects.pages():
+        ...     for obj in page:
+        ...         print(obj.key)
+
     A collection manager is not iterable. You **must** call one of the
     methods that return a :py:class:`ResourceCollection` before trying
     to iterate, slice, or convert to a list.
 
-    See :ref:`guide_collections` for a high-level overview of collections,
-    including when remote service requests are performed.
+    See the :ref:`guide_collections` guide for a high-level overview
+    of collections, including when remote service requests are performed.
 
     :type model: :py:class:`~boto3.resources.model.Collection`
     :param model: Collection model
@@ -235,6 +303,9 @@ class CollectionManager(object):
     :type service_model: :ref:`botocore.model.ServiceModel`
     :param service_model: The Botocore service model
     """
+    # The class to use when creating an iterator
+    _collection_cls = ResourceCollection
+
     def __init__(self, model, parent, factory, resource_defs,
                  service_model):
         self._model = model
@@ -262,8 +333,8 @@ def iterator(self, **kwargs):
         :rtype: :py:class:`ResourceCollection`
         :return: An iterable representing the collection of resources
         """
-        return ResourceCollection(self._model, self._parent,
-                                  self._handler, **kwargs)
+        return self._collection_cls(self._model, self._parent,
+                                    self._handler, **kwargs)
 
     # Set up some methods to proxy ResourceCollection methods
     def all(self, limit=None, page_size=None):
@@ -281,3 +352,82 @@ def limit(self, count):
     def page_size(self, count):
         return self.iterator(page_size=count)
     page_size.__doc__ = ResourceCollection.page_size.__doc__
+
+    def pages(self):
+        return self.iterator().pages()
+    pages.__doc__ = ResourceCollection.pages.__doc__
+
+
+class CollectionFactory(object):
+    """
+    A factory to create new
+    :py:class:`CollectionManager` and :py:class:`ResourceCollection`
+    subclasses from a :py:class:`~boto3.resources.model.Collection`
+    model. These subclasses include methods to perform batch operations.
+    """
+    def load_from_definition(self, service_name, resource_name,
+                             collection_name, model, resource_defs):
+        """
+        Loads a collection from a model, creating a new
+        :py:class:`CollectionManager` subclass
+        with the correct properties and methods, named based on the service
+        and resource name, e.g. ec2.InstanceCollectionManager. It also
+        creates a new :py:class:`ResourceCollection` subclass which is used
+        by the new manager class.
+
+        :type service_name: string
+        :param service_name: Name of the service to look up
+        :type resource_name: string
+        :param resource_name: Name of the resource to look up. For services,
+                              this should match the ``service_name``.
+        :type model: dict
+        :param model: The collection definition.
+        :type resource_defs: dict
+        :param resource_defs: The service's resource definitions, used to load
+                              collection resources (e.g. ``sqs.Queue``).
+        :rtype: Subclass of
+                :py:class:`CollectionManager`
+        :return: The collection class.
+        """
+        attrs = {}
+
+        self._load_batch_actions(attrs, model)
+
+        if service_name == resource_name:
+            cls_name = '{0}.{1}Collection'.format(
+                service_name, collection_name)
+        else:
+            cls_name = '{0}.{1}.{2}Collection'.format(
+                service_name, resource_name, collection_name)
+
+        collection_cls = type(str(cls_name), (ResourceCollection,),
+                              attrs)
+
+        attrs['_collection_cls'] = collection_cls
+        cls_name += 'Manager'
+
+        return type(str(cls_name), (CollectionManager,), attrs)
+
+    def _load_batch_actions(self, attrs, model):
+        """
+        Batch actions on the collection become methods on both
+        the collection manager and iterators.
+        """
+        for action_model in model.batch_actions:
+            snake_cased = xform_name(action_model.name)
+            attrs[snake_cased] = self._create_batch_action(
+                snake_cased, action_model)
+
+    def _create_batch_action(factory_self, snake_cased, action_model):
+        """
+        Creates a new method which makes a batch operation request
+        to the underlying service API.
+        """
+        action = BatchAction(action_model)
+
+        def batch_action(self, *args, **kwargs):
+            return action(self, *args, **kwargs)
+
+        batch_action.__name__ = str(snake_cased)
+        batch_action.__doc__ = 'TODO'
+        return batch_action
diff --git a/boto3/resources/factory.py b/boto3/resources/factory.py
index 5a1842e24c..da54a2b3a1 100644
--- a/boto3/resources/factory.py
+++ b/boto3/resources/factory.py
@@ -18,7 +18,7 @@
 
 from .action import ServiceAction
 from .base import ServiceResource
-from .collection import CollectionManager
+from .collection import CollectionFactory
 from .model import ResourceModel
 from .response import all_not_none, build_identifiers
 from ..exceptions import ResourceLoadException
@@ -35,6 +35,9 @@ class ResourceFactory(object):
     SQS resource) and another on models contained within the service (e.g. an
     SQS Queue resource).
     """
+    def __init__(self):
+        self._collection_factory = CollectionFactory()
+
     def load_from_definition(self, service_name, resource_name, model,
                              resource_defs, service_model):
         """
@@ -184,7 +187,8 @@ def _load_collections(self, attrs, model, resource_defs, service_model):
             snake_cased = self._check_allowed_name(
                 attrs, snake_cased, 'collection', model.name)
 
-            attrs[snake_cased] = self._create_collection(snake_cased,
+            attrs[snake_cased] = self._create_collection(
+                attrs['meta']['service_name'], model.name, snake_cased,
                 collection_model, resource_defs, service_model)
 
     def _load_references(self, attrs, service_name, resource_name,
@@ -267,14 +271,19 @@ def property_loader(self):
         property_loader.__doc__ = 'TODO'
         return property(property_loader)
 
-    def _create_collection(factory_self, snake_cased, collection_model,
+    def _create_collection(factory_self, service_name, resource_name,
+                           snake_cased, collection_model,
                            resource_defs, service_model):
         """
         Creates a new property on the resource to lazy-load a collection.
         """
+        cls = factory_self._collection_factory.load_from_definition(
+            service_name, resource_name, collection_model.name,
+            collection_model, resource_defs)
+
         def get_collection(self):
-            return CollectionManager(collection_model,
-                self, factory_self, resource_defs, service_model)
+            return cls(collection_model, self, factory_self,
+                       resource_defs, service_model)
 
         get_collection.__name__ = str(snake_cased)
         get_collection.__doc__ = 'TODO'
diff --git a/boto3/resources/model.py b/boto3/resources/model.py
index ef972c2b85..938e63d6ea 100644
--- a/boto3/resources/model.py
+++ b/boto3/resources/model.py
@@ -210,7 +210,16 @@ class Collection(Action):
     :type resource_defs: dict
     :param resource_defs: All resources defined in the service
     """
-    pass
+    @property
+    def batch_actions(self):
+        """
+        Get a list of batch actions supported by the resource type
+        contained in this action. This is a shortcut for accessing
+        the same information through the resource model.
+
+        :rtype: list(:py:class:`Action`)
+        """
+        return self.resource.model.batch_actions
 
 
 class SubResourceList(object):
diff --git a/boto3/resources/params.py b/boto3/resources/params.py
index 7a9a9fe92a..7d3c56c7f2 100644
--- a/boto3/resources/params.py
+++ b/boto3/resources/params.py
@@ -19,19 +19,27 @@
 INDEX_RE = re.compile('\[(.*)\]$')
 
 
-def create_request_parameters(parent, request_model):
+def create_request_parameters(parent, request_model, params=None):
     """
     Handle request parameters that can be filled in from identifiers,
     resource data members or constants.
 
+    By passing ``params``, you can invoke this method multiple times and
+    build up a parameter dict over time, which is particularly useful
+    for reverse JMESPath expressions that append to lists.
+
     :type parent: ServiceResource
     :param parent: The resource instance to which this action is attached.
     :type request_model: :py:class:`~boto3.resources.model.Request`
     :param request_model: The action request model.
+    :type params: dict
+    :param params: If set, then add to this existing dict. It is both
+                   edited in-place and returned.
     :rtype: dict
     :return: Pre-filled parameters to be sent to the request operation.
     """
-    params = {}
+    if params is None:
+        params = {}
 
     for param in request_model.params:
         source = param.source
diff --git a/docs/source/guide/collections.rst b/docs/source/guide/collections.rst
index db78401926..b6f03e450c 100644
--- a/docs/source/guide/collections.rst
+++ b/docs/source/guide/collections.rst
@@ -32,6 +32,10 @@ the following conditions:
 
       buckets = list(s3.buckets.all())
 
+* **Batch actions (see below)**::
+
+      s3.Bucket('my-bucket').objects.delete()
+
 Filtering
 ---------
 Some collections support extra arguments to filter the returned data set,
diff --git a/tests/unit/resources/test_action.py b/tests/unit/resources/test_action.py
index ad088b451d..f0e07e043e 100644
--- a/tests/unit/resources/test_action.py
+++ b/tests/unit/resources/test_action.py
@@ -11,7 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 
-from boto3.resources.action import ServiceAction
+from boto3.resources.action import BatchAction, ServiceAction
 from boto3.resources.model import Action
 from tests import BaseTestCase, mock
 
@@ -121,3 +121,117 @@ def test_service_action_calls_resource_handler(self, handler_mock, params_mock):
             service_model, action_model.resource,
             self.action_def['request']['operation'])
         handler_mock.return_value.assert_called_with(resource, {}, 'response')
+
+
+class TestBatchActionCall(BaseTestCase):
+    def setUp(self):
+        super(TestBatchActionCall, self).setUp()
+
+        self.action_def = {
+            'request': {
+                'operation': 'GetFrobs',
+                'params': []
+            }
+        }
+
+    @property
+    def model(self):
+        return Action('test', self.action_def, {})
+
+    def test_batch_action_gets_pages_from_collection(self):
+        collection = mock.Mock()
+        collection.pages.return_value = []
+        action = BatchAction(self.model)
+
+        action(collection)
+
+        collection.pages.assert_called_with()
+
+    def test_batch_action_creates_parameters_from_items(self):
+        self.action_def['request']['params'] = [
+            {'target': 'Bucket', 'sourceType': 'dataMember',
+             'source': 'BucketName'},
+            {'target': 'Delete.Objects[].Key', 'sourceType': 'dataMember',
+             'source': 'Key'}
+        ]
+
+        client = mock.Mock()
+
+        item1 = mock.Mock()
+        item1.meta = {
+            'service_name': 'test',
+            'client': client
+        }
+        item1.bucket_name = 'bucket'
+        item1.key = 'item1'
+
+        item2 = mock.Mock()
+        item2.bucket_name = 'bucket'
+        item2.key = 'item2'
+
+        collection = mock.Mock()
+        collection.pages.return_value = [[item1, item2]]
+
+        action = BatchAction(self.model)
+        action(collection)
+
+        client.get_frobs.assert_called_with(Bucket='bucket', Delete={
+            'Objects': [
+                {'Key': 'item1'},
+                {'Key': 'item2'}
+            ]
+        })
+
+    @mock.patch('boto3.resources.action.create_request_parameters',
+                return_value={})
+    def test_batch_action_skips_operation(self, crp_mock):
+        # In this test we have an item from the collection, but no
+        # parameters are set up. Because of this, we do NOT call
+        # the batch operation.
+        client = mock.Mock()
+
+        item = mock.Mock()
+        item.meta = {
+            'service_name': 'test',
+            'client': client
+        }
+
+        collection = mock.Mock()
+        collection.pages.return_value = [[item]]
+
+        model = self.model
+        action = BatchAction(model)
+        action(collection)
+
+        crp_mock.assert_called_with(item, model.request, params={})
+        client.get_frobs.assert_not_called()
+
+    @mock.patch('boto3.resources.action.create_request_parameters')
+    def test_batch_action_calls_operation(self, crp_mock):
+        # In this test we have an item and parameters, so the call
+        # to the batch operation should be made.
+        def side_effect(resource, model, params=None):
+            params['foo'] = 'bar'
+
+        crp_mock.side_effect = side_effect
+
+        client = mock.Mock()
+
+        item = mock.Mock()
+        item.meta = {
+            'service_name': 'test',
+            'client': client
+        }
+
+        collection = mock.Mock()
+        collection.pages.return_value = [[item]]
+
+        model = self.model
+        action = BatchAction(model)
+        action(collection)
+
+        # Here the call is made with params={}, but they are edited
+        # in-place so we need to compare to the final edited value.
+        crp_mock.assert_called_with(item, model.request,
+                                    params={'foo': 'bar'})
+        client.get_frobs.assert_called_with(foo='bar')
diff --git a/tests/unit/resources/test_collection.py b/tests/unit/resources/test_collection.py
index f87768abd0..7f7a116bc6 100644
--- a/tests/unit/resources/test_collection.py
+++ b/tests/unit/resources/test_collection.py
@@ -12,12 +12,108 @@
 # language governing permissions and limitations under the License.
 
 from botocore.model import ServiceModel
-from boto3.resources.collection import CollectionManager
+from boto3.resources.collection import CollectionFactory, CollectionManager, \
+                                       ResourceCollection
 from boto3.resources.factory import ResourceFactory
 from boto3.resources.model import Collection
 from tests import BaseTestCase, mock
 
 
+class TestCollectionFactory(BaseTestCase):
+    def setUp(self):
+        super(TestCollectionFactory, self).setUp()
+
+        self.client = mock.Mock()
+        self.client.can_paginate.return_value = False
+        meta = {
+            'client': self.client,
+            'service_name': 'test'
+        }
+        self.parent = mock.Mock()
+        self.parent.meta = meta
+        self.resource_factory = ResourceFactory()
+        self.service_model = ServiceModel({})
+
+        self.factory = CollectionFactory()
+        self.load = self.factory.load_from_definition
+
+    def test_create_subclasses(self):
+        resource_defs = {
+            'Frob': {},
+            'Chain': {
+                'hasMany': {
+                    'Frobs': {
+                        'request': {
+                            'operation': 'GetFrobs'
+                        },
+                        'resource': {
+                            'type': 'Frob'
+                        }
+                    }
+                }
+            }
+        }
+        collection_model = Collection(
+            'Frobs', resource_defs['Chain']['hasMany']['Frobs'],
+            resource_defs)
+
+        collection_cls = self.load('test', 'Chain', 'Frobs',
+                                   collection_model, resource_defs)
+
+        collection = collection_cls(
+            collection_model, self.parent, self.resource_factory,
+            resource_defs, self.service_model)
+
+        self.assertEqual(collection_cls.__name__,
+                        'test.Chain.FrobsCollectionManager')
+        self.assertIsInstance(collection, CollectionManager)
+
+        self.assertIsInstance(collection.all(), ResourceCollection)
+
+    @mock.patch('boto3.resources.collection.BatchAction')
+    def test_create_batch_actions(self, action_mock):
+        resource_defs = {
+            'Frob': {
+                'batchActions': {
+                    'Delete': {
+                        'request': {
+                            'operation': 'DeleteFrobs'
+                        }
+                    }
+                }
+            },
+            'Chain': {
+                'hasMany': {
+                    'Frobs': {
+                        'request': {
+                            'operation': 'GetFrobs'
+                        },
+                        'resource': {
+                            'type': 'Frob'
+                        }
+                    }
+                }
+            }
+        }
+
+        collection_model = Collection(
+            'Frobs', resource_defs['Chain']['hasMany']['Frobs'],
+            resource_defs)
+
+        collection_cls = self.load('test', 'Chain', 'Frobs',
+                                   collection_model, resource_defs)
+
+        collection = collection_cls(
+            collection_model, self.parent, self.resource_factory,
+            resource_defs, self.service_model)
+
+        self.assertTrue(hasattr(collection, 'delete'))
+
+        collection.delete()
+
+        action_mock.return_value.assert_called_with(collection)
+
+
 class TestResourceCollection(BaseTestCase):
     def setUp(self):
         super(TestResourceCollection, self).setUp()
@@ -195,6 +291,67 @@ def test_filters_non_paginated(self, handler):
         # Note - limit is not passed through to the low-level call
         self.client.get_frobs.assert_called_with(Param1='foo', Param2=3)
 
+    def test_page_iterator_returns_pages_of_items(self):
+        self.collection_def = {
+            'request': {
+                'operation': 'GetFrobs'
+            },
+            'resource': {
+                'type': 'Frob',
+                'identifiers': [
+                    {
+                        'target': 'Id',
+                        'sourceType': 'responsePath',
+                        'source': 'Frobs[].Id'
+                    }
+                ]
+            }
+        }
+        self.client.can_paginate.return_value = True
+        self.client.get_paginator.return_value.paginate.return_value = [
+            {
+                'Frobs': [
+                    {'Id': 'one'},
+                    {'Id': 'two'}
+                ]
+            }, {
+                'Frobs': [
+                    {'Id': 'three'},
+                    {'Id': 'four'}
+                ]
+            }
+        ]
+        collection = self.get_collection()
+        pages = list(collection.limit(3).pages())
+        self.assertEqual(len(pages), 2)
+        self.assertEqual(len(pages[0]), 2)
+        self.assertEqual(len(pages[1]), 1)
+
+    def test_page_iterator_page_size(self):
+        self.collection_def = {
+            'request': {
+                'operation': 'GetFrobs'
+            },
+            'resource': {
+                'type': 'Frob',
+                'identifiers': [
+                    {
+                        'target': 'Id',
+                        'sourceType': 'responsePath',
+                        'source': 'Frobs[].Id'
+                    }
+                ]
+            }
+        }
+        self.client.can_paginate.return_value = True
+        paginator = self.client.get_paginator.return_value
+        paginator.paginate.return_value = []
+
+        collection = self.get_collection()
+        list(collection.page_size(5).pages())
+
+        paginator.paginate.assert_called_with(page_size=5, max_items=None)
+
     def test_iteration_paginated(self):
         self.collection_def = {
             'request': {
diff --git a/tests/unit/resources/test_factory.py b/tests/unit/resources/test_factory.py
index 5c273f0a4e..2ed014049a 100644
--- a/tests/unit/resources/test_factory.py
+++ b/tests/unit/resources/test_factory.py
@@ -14,6 +14,7 @@
 from botocore.model import ServiceModel, StructureShape
 from boto3.exceptions import ResourceLoadException
 from boto3.resources.base import ServiceResource
+from boto3.resources.collection import CollectionManager
 from boto3.resources.factory import ResourceFactory
 from tests import BaseTestCase, mock
 
@@ -554,9 +555,8 @@ def test_resource_loads_references(self):
         self.assertIsInstance(resource.subnet, ServiceResource)
         self.assertEqual(resource.subnet.id, 'abc123')
 
-    @mock.patch('boto3.resources.factory.CollectionManager')
     @mock.patch('boto3.resources.model.Collection')
-    def test_resource_loads_collections(self, mock_model, collection_cls):
+    def test_resource_loads_collections(self, mock_model):
         model = {
             'hasMany': {
                 u'Queues': {
@@ -579,8 +579,5 @@ def test_resource_loads_collections(self, mock_model, collection_cls):
 
         self.assertTrue(hasattr(resource, 'queues'),
             'Resource should expose queues collection')
-        self.assertEqual(resource.queues, collection_cls.return_value,
+        self.assertIsInstance(resource.queues, CollectionManager,
             'Queues collection should be a collection manager')
-
-        collection_cls.assert_called_with(mock_model.return_value,
-            resource, self.factory, defs, service_model)
diff --git a/tests/unit/resources/test_params.py b/tests/unit/resources/test_params.py
index 0202cb0cbd..f7774304e8 100644
--- a/tests/unit/resources/test_params.py
+++ b/tests/unit/resources/test_params.py
@@ -103,7 +103,7 @@ def test_service_action_params_invalid(self):
         with self.assertRaises(NotImplementedError):
             create_request_parameters(None, request_model)
 
-    def test_action_params_list(self):
+    def test_service_action_params_list(self):
         request_model = Request({
             'operation': 'GetFrobs',
             'params': [
@@ -124,6 +124,38 @@ def test_action_params_list(self):
         self.assertIn('w-url', params['WarehouseUrls'],
             'Parameter not in expected list')
 
+    def test_service_action_params_reuse(self):
+        request_model = Request({
+            'operation': 'GetFrobs',
+            'params': [
+                {
+                    'target': 'Delete.Objects[].Key',
+                    'sourceType': 'dataMember',
+                    'source': 'Key'
+                }
+            ]
+        })
+
+        item1 = mock.Mock()
+        item1.key = 'item1'
+
+        item2 = mock.Mock()
+        item2.key = 'item2'
+
+        # Here we create params and then re-use it to build up a more
+        # complex structure over multiple calls.
+        params = create_request_parameters(item1, request_model)
+        create_request_parameters(item2, request_model, params=params)
+
+        self.assertEqual(params, {
+            'Delete': {
+                'Objects': [
+                    {'Key': 'item1'},
+                    {'Key': 'item2'}
+                ]
+            }
+        })
+
 
 class TestStructBuilder(BaseTestCase):
     def test_simple_value(self):