docs: 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)