'Perform an aggregation using the aggregation framework on this
collection.
All optional aggregate parameters should be passed as keyword arguments
to this method. Valid options include, but are not limited to:
- `allowDiskUse` (bool): Enables writing to temporary files. When set
to True, aggregation stages can write data to the _tmp subdirectory
of the --dbpath directory. The default is False.
- `maxTimeMS` (int): The maximum amount of time to allow the operation
to run in milliseconds.
- `batchSize` (int): The maximum number of documents to return per
batch. Ignored if the connected mongod or mongos does not support
returning aggregate results using a cursor, or `useCursor` is
``False``.
- `useCursor` (bool): Requests that the `server` provide results
using a cursor, if possible. Ignored if the connected mongod or
mongos does not support returning aggregate results using a cursor.
The default is ``True``. Set this to ``False`` when upgrading a 2.4
or older sharded cluster to 2.6 or newer (see the warning below).
- `collation` (optional): An instance of
:class:`~pymongo.collation.Collation`. This option is only supported
on MongoDB 3.4 and above.
The :meth:`aggregate` method obeys the :attr:`read_preference` of this
:class:`Collection`. Please note that using the ``$out`` pipeline stage
requires a read preference of
:attr:`~pymongo.read_preferences.ReadPreference.PRIMARY` (the default).
The server will raise an error if the ``$out`` pipeline stage is used
with any other read preference.
.. warning:: When upgrading a 2.4 or older sharded cluster to 2.6 or
newer the `useCursor` option **must** be set to ``False``
until all shards have been upgraded to 2.6 or newer.
.. note:: This method does not support the \'explain\' option. Please
use :meth:`~pymongo.database.Database.command` instead. An
example is included in the :ref:`aggregate-examples` documentation.
.. note:: The :attr:`~pymongo.collection.Collection.write_concern` of
this collection is automatically applied to this operation when using
MongoDB >= 3.4.
:Parameters:
- `pipeline`: a list of aggregation pipeline stages
- `**kwargs` (optional): See list of options above.
:Returns:
A :class:`~pymongo.command_cursor.CommandCursor` over the result
set.
.. versionchanged:: 3.4
Apply this collection\'s write concern automatically to this operation
when connected to MongoDB >= 3.4. Support the `collation` option.
.. versionchanged:: 3.0
The :meth:`aggregate` method always returns a CommandCursor. The
pipeline argument must be a list.
.. versionchanged:: 2.7
When the cursor option is used, return
:class:`~pymongo.command_cursor.CommandCursor` instead of
:class:`~pymongo.cursor.Cursor`.
.. versionchanged:: 2.6
Added cursor support.
.. versionadded:: 2.3
.. seealso:: :doc:`/examples/aggregation`
.. _aggregate command:
http://docs.mongodb.org/manual/applications/aggregation'

def aggregate(self, pipeline, **kwargs):

    if (not isinstance(pipeline, list)):
        raise TypeError('pipeline    must    be    a    list')
    if ('explain' in kwargs):
        raise ConfigurationError('The    explain    option    is    not    supported.    Use    Database.command    instead.')
    collation = validate_collation_or_none(kwargs.pop('collation', None))
    cmd = SON([('aggregate', self.__name), ('pipeline', pipeline)])
    batch_size = common.validate_positive_integer_or_none('batchSize', kwargs.pop('batchSize', None))
    use_cursor = common.validate_boolean('useCursor', kwargs.pop('useCursor', True))
    with self._socket_for_reads() as (sock_info, slave_ok):
        if (sock_info.max_wire_version > 0):
            if use_cursor:
                if ('cursor' not in kwargs):
                    kwargs['cursor'] = {}
                if (batch_size is not None):
                    kwargs['cursor']['batchSize'] = batch_size
        dollar_out = (pipeline and ('$out' in pipeline[(-1)]))
        if ((sock_info.max_wire_version >= 5) and dollar_out and self.write_concern):
            cmd['writeConcern'] = self.write_concern.document
        cmd.update(kwargs)
        if ((sock_info.max_wire_version >= 4) and ('readConcern' not in cmd)):
            if dollar_out:
                result = self._command(sock_info, cmd, slave_ok, parse_write_concern_error=True, collation=collation)
            else:
                result = self._command(sock_info, cmd, slave_ok, read_concern=self.read_concern, collation=collation)
        else:
            result = self._command(sock_info, cmd, slave_ok, parse_write_concern_error=dollar_out, collation=collation)
        if ('cursor' in result):
            cursor = result['cursor']
        else:
            cursor = {'id': 0, 'firstBatch': result['result'], 'ns': self.full_name}
        return CommandCursor(self, cursor, sock_info.address).batch_size((batch_size or 0))
