(self, pipeline, **kwargs)
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
Perform an aggregation using the aggregation framework on this
        collection.
Perform an aggregation using the aggregation framework on this collection .
def aggregate(self, pipeline, **kwargs):
        """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
        """
        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)])

        # Remove things that are not command options.
        batch_size = common.validate_positive_integer_or_none(
            "batchSize", kwargs.pop("batchSize", None))
        use_cursor = common.validate_boolean(
            "useCursor", kwargs.pop("useCursor", True))
        # If the server does not support the "cursor" option we
        # ignore useCursor and batchSize.
        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)

            # Apply this Collection's read concern if $out is not in the
            # pipeline.
            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:
                # Pre-MongoDB 2.6. Fake a cursor.
                cursor = {
                    "id": 0,
                    "firstBatch": result["result"],
                    "ns": self.full_name,
                }
            return CommandCursor(
                self, cursor, sock_info.address).batch_size(batch_size or 0)
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 ) ] ) # Remove things that are not command options. batch_size = common . validate_positive_integer_or_none ( "batchSize" , kwargs . pop ( "batchSize" , None ) ) use_cursor = common . validate_boolean ( "useCursor" , kwargs . pop ( "useCursor" , True ) ) # If the server does not support the "cursor" option we # ignore useCursor and batchSize. 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 ) # Apply this Collection's read concern if $out is not in the # pipeline. 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 : # Pre-MongoDB 2.6. Fake a cursor. cursor = { "id" : 0 , "firstBatch" : result [ "result" ] , "ns" : self . full_name , } return CommandCursor ( self , cursor , sock_info . address ) . batch_size ( batch_size or 0 )