def group(self, key, condition, initial, reduce, finalize=None):
        """Perform a query similar to an SQL *group by* operation.

        Returns an array of grouped items.

        The `key` parameter can be:

          - ``None`` to use the entire document as a key.
          - A :class:`list` of keys (each a :class:`basestring`
            (:class:`str` in python 3)) to group by.
          - A :class:`basestring` (:class:`str` in python 3), or
            :class:`~bson.code.Code` instance containing a JavaScript
            function to be applied to each document, returning the key
            to group by.

        With :class:`~pymongo.replica_set_connection.ReplicaSetConnection`
        or :class:`~pymongo.master_slave_connection.MasterSlaveConnection`,
        if the `read_preference` attribute of this instance is not set to
        :attr:`pymongo.ReadPreference.PRIMARY` or the (deprecated)
        `slave_okay` attribute of this instance is set to `True` the group
        command will be sent to a secondary or slave.

        :Parameters:
          - `key`: fields to group by (see above description)
          - `condition`: specification of rows to be
            considered (as a :meth:`find` query specification)
          - `initial`: initial value of the aggregation counter object
          - `reduce`: aggregation function as a JavaScript string
          - `finalize`: function to be called on each object in output list.

        .. versionchanged:: 2.2
           Removed deprecated argument: command

        .. versionchanged:: 1.4
           The `key` argument can now be ``None`` or a JavaScript function,
           in addition to a :class:`list` of keys.

        .. versionchanged:: 1.3
           The `command` argument now defaults to ``True`` and is deprecated.
        """

        group = {}
        if isinstance(key, basestring):
            group["$keyf"] = Code(key)
        elif key is not None:
            group = {"key": helpers._fields_list_to_dict(key)}
        group["ns"] = self.__name
        group["$reduce"] = Code(reduce)
        group["cond"] = condition
        group["initial"] = initial
        if finalize is not None:
            group["finalize"] = Code(finalize)

        use_master = not self.slave_okay and not self.read_preference

        return self.__database.command("group", group,
                                       uuid_subtype=self.__uuid_subtype,
                                       read_preference=self.read_preference,
                                       slave_okay=self.slave_okay,
                                       _use_master=use_master)["retval"]

    def __init__(self, collection, spec=None, fields=None, skip=0, limit=0,
                 timeout=True, snapshot=False, tailable=False, sort=None,
                 max_scan=None, as_class=None,
                 _sock=None, _must_use_master=False, _is_command=False):
        """Create a new cursor.

        Should not be called directly by application developers - see
        :meth:`~pymongo.collection.Collection.find` instead.

        .. mongodoc:: cursors
        """
        self.__id = None

        if spec is None:
            spec = {}

        if not isinstance(spec, dict):
            raise TypeError("spec must be an instance of dict")
        if not isinstance(skip, int):
            raise TypeError("skip must be an instance of int")
        if not isinstance(limit, int):
            raise TypeError("limit must be an instance of int")
        if not isinstance(timeout, bool):
            raise TypeError("timeout must be an instance of bool")
        if not isinstance(snapshot, bool):
            raise TypeError("snapshot must be an instance of bool")
        if not isinstance(tailable, bool):
            raise TypeError("tailable must be an instance of bool")

        if fields is not None:
            if not fields:
                fields = {"_id": 1}
            if not isinstance(fields, dict):
                fields = helpers._fields_list_to_dict(fields)

        if as_class is None:
            as_class = collection.database.connection.document_class

        self.__collection = collection
        self.__spec = spec
        self.__fields = fields
        self.__skip = skip
        self.__limit = limit
        self.__timeout = timeout
        self.__tailable = tailable
        self.__snapshot = snapshot
        self.__ordering = sort and helpers._index_document(sort) or None
        self.__max_scan = max_scan
        self.__explain = False
        self.__hint = None
        self.__as_class = as_class
        self.__socket = _sock
        self.__must_use_master = _must_use_master
        self.__is_command = _is_command

        self.__data = []
        self.__connection_id = None
        self.__retrieved = 0
        self.__killed = False

    def _command(self, command, value=1,
                 check=True, allowable_errors=None,
                 uuid_subtype=OLD_UUID_SUBTYPE, compile_re=True,
                 read_preference=None, **kwargs):
        """Internal command helper.
        """

        if isinstance(command, basestring):
            command_name = command.lower()
            command = SON([(command, value)])
        else:
            command_name = command.keys()[0].lower()

        as_class = kwargs.pop('as_class', None)
        fields = kwargs.pop('fields', None)
        if fields is not None and not isinstance(fields, dict):
            fields = helpers._fields_list_to_dict(fields)
        command.update(kwargs)

        orig = mode = read_preference or self.read_preference
        if command_name not in SECONDARY_OK_COMMANDS:
            mode = ReadPreference.PRIMARY

        # Special-case: mapreduce can go to secondaries only if inline
        elif command_name == 'mapreduce':
            out = command.get('out')
            if not isinstance(out, dict) or not out.get('inline'):
                mode = ReadPreference.PRIMARY

        # Special-case: aggregate with $out cannot go to secondaries.
        elif command_name == 'aggregate':
            for stage in command.get('pipeline', []):
                if '$out' in stage:
                    mode = ReadPreference.PRIMARY
                    break

        # Warn if mode will override read_preference.
        if mode != orig:
            warnings.warn("%s does not support %s read preference "
                          "and will be routed to the primary instead." %
                          (command_name, orig.name), UserWarning)


        cursor = self["$cmd"].find(command,
                                   fields=fields,
                                   limit=-1,
                                   as_class=as_class,
                                   read_preference=mode,
                                   compile_re=compile_re,
                                   _uuid_subtype=uuid_subtype)
        for doc in cursor:
            result = doc

        if check:
            msg = "command %s failed: %%s" % repr(command).replace("%", "%%")
            helpers._check_command_response(result, self.connection.disconnect,
                                            msg, allowable_errors)

        return result, cursor.conn_id

    def group(self, key, condition, initial, reduce, finalize=None,
              command=True):
        """Perform a query similar to an SQL *group by* operation.

        Returns an array of grouped items.

        The `key` parameter can be:

          - ``None`` to use the entire document as a key.
          - A :class:`list` of keys (each a :class:`basestring`) to group by.
          - A :class:`basestring` or :class:`~pymongo.code.Code` instance
            containing a JavaScript function to be applied to each document,
            returning the key to group by.

        :Parameters:
          - `key`: fields to group by (see above description)
          - `condition`: specification of rows to be
            considered (as a :meth:`find` query specification)
          - `initial`: initial value of the aggregation counter object
          - `reduce`: aggregation function as a JavaScript string
          - `finalize`: function to be called on each object in output list.
          - `command` (optional): DEPRECATED if ``True``, run the group as a
            command instead of in an eval - this option is deprecated and
            will be removed in favor of running all groups as commands

        .. versionchanged:: 1.4
           The `key` argument can now be ``None`` or a JavaScript function,
           in addition to a :class:`list` of keys.
        .. versionchanged:: 1.3
           The `command` argument now defaults to ``True`` and is deprecated.
        """
        if not command:
            warnings.warn("eval-based groups are deprecated, and the "
                          "command option will be removed.",
                          DeprecationWarning)

        group = {}
        if isinstance(key, basestring):
            group["$keyf"] = Code(key)
        elif key is not None:
            group = {"key": helpers._fields_list_to_dict(key)}
        group["ns"] = self.__name
        group["$reduce"] = Code(reduce)
        group["cond"] = condition
        group["initial"] = initial
        if finalize is not None:
            group["finalize"] = Code(finalize)

        return self.__database.command("group", group)["retval"]

    def add_handler(self,collection,spec,fields,skip,limit,callback,_must_use_master=False,_is_command=False):
    
        if fields is not None:
            if not fields:
                fields = ["_id"]
            fields = helpers._fields_list_to_dict(fields)
    
        slave_okay = collection.database.connection.slave_okay
        timeout = True
        tailable = False
        
        io_loop = self.settings['io_loop']
    
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        IOStream = tornado.iostream.IOStream(sock,io_loop)
        self.__IOStream = IOStream
        
                   
        self.cursor = Cursor(collection,
                        spec,
                        fields,
                        skip,
                        limit,
                        slave_okay, 
                        timeout,
                        tailable,
                        _IOStream=IOStream,
                        _must_use_master=_must_use_master, 
                        _is_command=_is_command,
                        sort = self.__ordering.items() if self.__ordering else None
                        )


        callback = functools.partial(callback,self)
        
        start_callback = functools.partial(start,self,callback)

        IOStream.connect(('localhost',27017),callback=start_callback)

    def __init__(self, collection, spec=None, fields=None, skip=0, limit=0,
                 timeout=True, snapshot=False, tailable=False, sort=None,
                 max_scan=None, as_class=None, slave_okay=False,
                 await_data=False, partial=False, manipulate=True,
                 read_preference=ReadPreference.PRIMARY, tag_sets=[{}],
                 secondary_acceptable_latency_ms=None,
                 _must_use_master=False, _uuid_subtype=None, **kwargs):
        """Create a new cursor.

        Should not be called directly by application developers - see
        :meth:`~pymongo.collection.Collection.find` instead.

        .. mongodoc:: cursors
        """
        self.__id = None

        if spec is None:
            spec = {}

        if not isinstance(spec, dict):
            raise TypeError("spec must be an instance of dict")
        if not isinstance(skip, int):
            raise TypeError("skip must be an instance of int")
        if not isinstance(limit, int):
            raise TypeError("limit must be an instance of int")
        if not isinstance(timeout, bool):
            raise TypeError("timeout must be an instance of bool")
        if not isinstance(snapshot, bool):
            raise TypeError("snapshot must be an instance of bool")
        if not isinstance(tailable, bool):
            raise TypeError("tailable must be an instance of bool")
        if not isinstance(slave_okay, bool):
            raise TypeError("slave_okay must be an instance of bool")
        if not isinstance(await_data, bool):
            raise TypeError("await_data must be an instance of bool")
        if not isinstance(partial, bool):
            raise TypeError("partial must be an instance of bool")

        if fields is not None:
            if not fields:
                fields = {"_id": 1}
            if not isinstance(fields, dict):
                fields = helpers._fields_list_to_dict(fields)

        if as_class is None:
            as_class = collection.database.connection.document_class

        self.__collection = collection
        self.__spec = spec
        self.__fields = fields
        self.__skip = skip
        self.__limit = limit
        self.__batch_size = 0

        # This is ugly. People want to be able to do cursor[5:5] and
        # get an empty result set (old behavior was an
        # exception). It's hard to do that right, though, because the
        # server uses limit(0) to mean 'no limit'. So we set __empty
        # in that case and check for it when iterating. We also unset
        # it anytime we change __limit.
        self.__empty = False

        self.__timeout = timeout
        self.__tailable = tailable
        self.__await_data = tailable and await_data
        self.__partial = partial
        self.__snapshot = snapshot
        self.__ordering = sort and helpers._index_document(sort) or None
        self.__max_scan = max_scan
        self.__explain = False
        self.__hint = None
        self.__as_class = as_class
        self.__slave_okay = slave_okay
        self.__manipulate = manipulate
        self.__read_preference = read_preference
        self.__tag_sets = tag_sets
        self.__secondary_acceptable_latency_ms = secondary_acceptable_latency_ms
        self.__tz_aware = collection.database.connection.tz_aware
        self.__must_use_master = _must_use_master
        self.__uuid_subtype = _uuid_subtype or collection.uuid_subtype
        self.__query_flags = 0

        self.__data = deque()
        self.__connection_id = None
        self.__retrieved = 0
        self.__killed = False

        # this is for passing network_timeout through if it's specified
        # need to use kwargs as None is a legit value for network_timeout
        self.__kwargs = kwargs

#.........这里部分代码省略.........
        >>> db.command("buildinfo")

        For a command where the value matters, like ``{collstats:
        collection_name}`` we can do:

        >>> db.command("collstats", collection_name)

        For commands that take additional arguments we can use
        kwargs. So ``{filemd5: object_id, root: file_root}`` becomes:

        >>> db.command("filemd5", object_id, root=file_root)

        :Parameters:
          - `command`: document representing the command to be issued,
            or the name of the command (for simple commands only).

            .. note:: the order of keys in the `command` document is
               significant (the "verb" must come first), so commands
               which require multiple keys (e.g. `findandmodify`)
               should use an instance of :class:`~bson.son.SON` or
               a string and kwargs instead of a Python `dict`.

          - `value` (optional): value to use for the command verb when
            `command` is passed as a string
          - `check` (optional): check the response for errors, raising
            :class:`~pymongo.errors.OperationFailure` if there are any
          - `allowable_errors`: if `check` is ``True``, error messages
            in this list will be ignored by error-checking
          - `uuid_subtype` (optional): The BSON binary subtype to use
            for a UUID used in this command.
          - `read_preference`: The read preference for this connection.
            See :class:`~pymongo.read_preferences.ReadPreference` for available
            options.
          - `tag_sets`: Read from replica-set members with these tags.
            To specify a priority-order for tag sets, provide a list of
            tag sets: ``[{'dc': 'ny'}, {'dc': 'la'}, {}]``. A final, empty tag
            set, ``{}``, means "read from any member that matches the mode,
            ignoring tags." ReplicaSetConnection tries each set of tags in turn
            until it finds a set of tags with at least one matching member.
          - `secondary_acceptable_latency_ms`: Any replica-set member whose
            ping time is within secondary_acceptable_latency_ms of the nearest
            member may accept reads. Default 15 milliseconds.
          - `**kwargs` (optional): additional keyword arguments will
            be added to the command document before it is sent

        .. versionchanged:: 2.3
           Added `tag_sets` and `secondary_acceptable_latency_ms` options.
        .. versionchanged:: 2.2
           Added support for `as_class` - the class you want to use for
           the resulting documents
        .. versionchanged:: 1.6
           Added the `value` argument for string commands, and keyword
           arguments for additional command options.
        .. versionchanged:: 1.5
           `command` can be a string in addition to a full document.
        .. versionadded:: 1.4

        .. mongodoc:: commands
        """

        if isinstance(command, basestring):
            command = SON([(command, value)])

        command_name = command.keys()[0].lower()
        must_use_master = kwargs.pop("_use_master", False)
        if command_name not in rp.secondary_ok_commands:
            must_use_master = True

        # Special-case: mapreduce can go to secondaries only if inline
        if command_name == "mapreduce":
            out = command.get("out") or kwargs.get("out")
            if not isinstance(out, dict) or not out.get("inline"):
                must_use_master = True

        extra_opts = {
            "as_class": kwargs.pop("as_class", None),
            "slave_okay": kwargs.pop("slave_okay", self.slave_okay),
            "_must_use_master": must_use_master,
            "_uuid_subtype": uuid_subtype,
        }

        extra_opts["read_preference"] = kwargs.pop("read_preference", self.read_preference)
        extra_opts["tag_sets"] = kwargs.pop("tag_sets", self.tag_sets)
        extra_opts["secondary_acceptable_latency_ms"] = kwargs.pop(
            "secondary_acceptable_latency_ms", self.secondary_acceptable_latency_ms
        )

        fields = kwargs.get("fields")
        if fields is not None and not isinstance(fields, dict):
            kwargs["fields"] = helpers._fields_list_to_dict(fields)

        command.update(kwargs)

        result = self["$cmd"].find_one(command, **extra_opts)

        if check:
            msg = "command %s failed: %%s" % repr(command).replace("%", "%%")
            helpers._check_command_response(result, self.connection.disconnect, msg, allowable_errors)

        return result

    def _command(self, command, value=1,
                 check=True, allowable_errors=None,
                 uuid_subtype=OLD_UUID_SUBTYPE, compile_re=True, **kwargs):
        """Internal command helper.
        """

        if isinstance(command, str):
            command = SON([(command, value)])

        command_name = list(command.keys())[0].lower()
        must_use_master = kwargs.pop('_use_master', False)
        if command_name not in secondary_ok_commands:
            must_use_master = True

        # Special-case: mapreduce can go to secondaries only if inline
        if command_name == 'mapreduce':
            out = command.get('out') or kwargs.get('out')
            if not isinstance(out, dict) or not out.get('inline'):
                must_use_master = True

        # Special-case: aggregate with $out cannot go to secondaries.
        if command_name == 'aggregate':
            for stage in kwargs.get('pipeline', []):
                if '$out' in stage:
                    must_use_master = True
                    break

        extra_opts = {
            'as_class': kwargs.pop('as_class', None),
            'slave_okay': kwargs.pop('slave_okay', self.slave_okay),
            '_must_use_master': must_use_master,
            '_uuid_subtype': uuid_subtype
        }

        extra_opts['read_preference'] = kwargs.pop(
            'read_preference',
            self.read_preference)
        extra_opts['tag_sets'] = kwargs.pop(
            'tag_sets',
            self.tag_sets)
        extra_opts['secondary_acceptable_latency_ms'] = kwargs.pop(
            'secondary_acceptable_latency_ms',
            self.secondary_acceptable_latency_ms)
        extra_opts['compile_re'] = compile_re

        fields = kwargs.get('fields')
        if fields is not None and not isinstance(fields, dict):
            kwargs['fields'] = helpers._fields_list_to_dict(fields)

        command.update(kwargs)

        # Warn if must_use_master will override read_preference.
        if (extra_opts['read_preference'] != ReadPreference.PRIMARY and
                extra_opts['_must_use_master']):
            warnings.warn("%s does not support %s read preference "
                          "and will be routed to the primary instead." %
                          (command_name,
                           modes[extra_opts['read_preference']]),
                          UserWarning, stacklevel=3)

        cursor = self["$cmd"].find(command, **extra_opts).limit(-1)
        for doc in cursor:
            result = doc

        if check:
            msg = "command %s failed: %%s" % repr(command).replace("%", "%%")
            helpers._check_command_response(result, self.connection.disconnect,
                                            msg, allowable_errors)

        return result, cursor.conn_id

    def command(self, command, value=1,
                check=True, allowable_errors=[],
                uuid_subtype=UUID_SUBTYPE, **kwargs):
        """Issue a MongoDB command.

        Send command `command` to the database and return the
        response. If `command` is an instance of :class:`basestring`
        then the command {`command`: `value`} will be sent. Otherwise,
        `command` must be an instance of :class:`dict` and will be
        sent as is.

        Any additional keyword arguments will be added to the final
        command document before it is sent.

        For example, a command like ``{buildinfo: 1}`` can be sent
        using:

        >>> db.command("buildinfo")

        For a command where the value matters, like ``{collstats:
        collection_name}`` we can do:

        >>> db.command("collstats", collection_name)

        For commands that take additional arguments we can use
        kwargs. So ``{filemd5: object_id, root: file_root}`` becomes:

        >>> db.command("filemd5", object_id, root=file_root)

        :Parameters:
          - `command`: document representing the command to be issued,
            or the name of the command (for simple commands only).

            .. note:: the order of keys in the `command` document is
               significant (the "verb" must come first), so commands
               which require multiple keys (e.g. `findandmodify`)
               should use an instance of :class:`~bson.son.SON` or
               a string and kwargs instead of a Python `dict`.

          - `value` (optional): value to use for the command verb when
            `command` is passed as a string
          - `check` (optional): check the response for errors, raising
            :class:`~pymongo.errors.OperationFailure` if there are any
          - `allowable_errors`: if `check` is ``True``, error messages
            in this list will be ignored by error-checking
          - `uuid_subtype` (optional): The BSON binary subtype to use
            for a UUID used in this command.
          - `**kwargs` (optional): additional keyword arguments will
            be added to the command document before it is sent

        .. versionchanged:: 1.6
           Added the `value` argument for string commands, and keyword
           arguments for additional command options.
        .. versionchanged:: 1.5
           `command` can be a string in addition to a full document.
        .. versionadded:: 1.4

        .. mongodoc:: commands
        """

        if isinstance(command, basestring):
            command = SON([(command, value)])

        use_master = kwargs.pop('_use_master', True)

        fields = kwargs.get('fields')
        if fields is not None and not isinstance(fields, dict):
                kwargs['fields'] = helpers._fields_list_to_dict(fields)

        command.update(kwargs)

        result = self["$cmd"].find_one(command,
                                       _must_use_master=use_master,
                                       _is_command=True,
                                       _uuid_subtype = uuid_subtype)

        if check:
            msg = "command %s failed: %%s" % repr(command).replace("%", "%%")
            helpers._check_command_response(result, self.connection.disconnect,
                                            msg, allowable_errors)

        return result

    def __init__(self, collection, filter=None, projection=None, skip=0,
                 limit=0, no_cursor_timeout=False,
                 cursor_type=CursorType.NON_TAILABLE,
                 sort=None, allow_partial_results=False, oplog_replay=False,
                 modifiers=None, batch_size=0, manipulate=True,
                 collation=None, hint=None, max_scan=None, max_time_ms=None,
                 max=None, min=None, return_key=False, show_record_id=False,
                 snapshot=False, comment=None, session=None):
        """Create a new cursor.

        Should not be called directly by application developers - see
        :meth:`~pymongo.collection.Collection.find` instead.

        .. mongodoc:: cursors
        """
        # Initialize all attributes used in __del__ before possibly raising
        # an error to avoid attribute errors during garbage collection.
        self.__id = None
        self.__exhaust = False
        self.__exhaust_mgr = None
        self.__killed = False

        if session:
            self.__session = session
            self.__explicit_session = True
        else:
            self.__session = None
            self.__explicit_session = False

        spec = filter
        if spec is None:
            spec = {}

        validate_is_mapping("filter", spec)
        if not isinstance(skip, int):
            raise TypeError("skip must be an instance of int")
        if not isinstance(limit, int):
            raise TypeError("limit must be an instance of int")
        validate_boolean("no_cursor_timeout", no_cursor_timeout)
        if cursor_type not in (CursorType.NON_TAILABLE, CursorType.TAILABLE,
                               CursorType.TAILABLE_AWAIT, CursorType.EXHAUST):
            raise ValueError("not a valid value for cursor_type")
        validate_boolean("allow_partial_results", allow_partial_results)
        validate_boolean("oplog_replay", oplog_replay)
        if modifiers is not None:
            warnings.warn("the 'modifiers' parameter is deprecated",
                          DeprecationWarning, stacklevel=2)
            validate_is_mapping("modifiers", modifiers)
        if not isinstance(batch_size, integer_types):
            raise TypeError("batch_size must be an integer")
        if batch_size < 0:
            raise ValueError("batch_size must be >= 0")

        if projection is not None:
            if not projection:
                projection = {"_id": 1}
            projection = helpers._fields_list_to_dict(projection, "projection")

        self.__collection = collection
        self.__spec = spec
        self.__projection = projection
        self.__skip = skip
        self.__limit = limit
        self.__batch_size = batch_size
        self.__modifiers = modifiers and modifiers.copy() or {}
        self.__ordering = sort and helpers._index_document(sort) or None
        self.__max_scan = max_scan
        self.__explain = False
        self.__comment = comment
        self.__max_time_ms = max_time_ms
        self.__max_await_time_ms = None
        self.__max = max
        self.__min = min
        self.__manipulate = manipulate
        self.__collation = validate_collation_or_none(collation)
        self.__return_key = return_key
        self.__show_record_id = show_record_id
        self.__snapshot = snapshot
        self.__set_hint(hint)

        # Exhaust cursor support
        if cursor_type == CursorType.EXHAUST:
            if self.__collection.database.client.is_mongos:
                raise InvalidOperation('Exhaust cursors are '
                                       'not supported by mongos')
            if limit:
                raise InvalidOperation("Can't use limit and exhaust together.")
            self.__exhaust = True

        # This is ugly. People want to be able to do cursor[5:5] and
        # get an empty result set (old behavior was an
        # exception). It's hard to do that right, though, because the
        # server uses limit(0) to mean 'no limit'. So we set __empty
        # in that case and check for it when iterating. We also unset
        # it anytime we change __limit.
        self.__empty = False

        self.__data = deque()
        self.__address = None
        self.__retrieved = 0
#.........这里部分代码省略.........

    def _command(self, command, value=1,
                 check=True, allowable_errors=None,
                 uuid_subtype=OLD_UUID_SUBTYPE, compile_re=True,
                 read_preference=None, codec_options=None, **kwargs):
        """Internal command helper.
        """

        if isinstance(command, basestring):
            command = SON([(command, value)])

        command_name = command.keys()[0].lower()
        must_use_master = kwargs.pop('_use_master', False)
        if command_name not in secondary_ok_commands:
            must_use_master = True

        # Special-case: mapreduce can go to secondaries only if inline
        if command_name == 'mapreduce':
            out = command.get('out') or kwargs.get('out')
            if not isinstance(out, dict) or not out.get('inline'):
                must_use_master = True

        # Special-case: aggregate with $out cannot go to secondaries.
        if command_name == 'aggregate':
            for stage in kwargs.get('pipeline', []):
                if '$out' in stage:
                    must_use_master = True
                    break

        if codec_options is None or 'as_class' in kwargs:
            opts = {}
            if 'as_class' in kwargs:
                opts['document_class'] = kwargs.pop('as_class')
            # 'as_class' must be in kwargs so don't use document_class
            if codec_options:
                opts['tz_aware'] = codec_options.tz_aware
                opts['uuid_representation'] = codec_options.uuid_representation
            else:
                opts['uuid_representation'] = uuid_subtype
            codec_options = _CodecOptions(**opts)

        extra_opts = {
            'slave_okay': kwargs.pop('slave_okay', self.slave_okay),
            '_codec_options': codec_options,
            '_must_use_master': must_use_master,
        }

        if isinstance(read_preference, _ServerMode):
            extra_opts['read_preference'] = read_preference.mode
            extra_opts['tag_sets'] = read_preference.tag_sets
        else:
            if read_preference is None:
                read_preference = self.read_preference
            extra_opts['read_preference'] = read_preference
            extra_opts['tag_sets'] = kwargs.pop(
                'tag_sets',
                self.tag_sets)
        extra_opts['secondary_acceptable_latency_ms'] = kwargs.pop(
            'secondary_acceptable_latency_ms',
            self.secondary_acceptable_latency_ms)
        extra_opts['compile_re'] = compile_re

        fields = kwargs.get('fields')
        if fields is not None and not isinstance(fields, dict):
            kwargs['fields'] = helpers._fields_list_to_dict(fields)

        command.update(kwargs)

        # Warn if must_use_master will override read_preference.
        if (extra_opts['read_preference'] != ReadPreference.PRIMARY and
                extra_opts['_must_use_master'] and self.connection._rs_client):
            warnings.warn("%s does not support %s read preference "
                          "and will be routed to the primary instead." %
                          (command_name,
                           modes[extra_opts['read_preference']]),
                          UserWarning, stacklevel=3)

        cursor = self["$cmd"].find(command, **extra_opts).limit(-1)
        for doc in cursor:
            result = doc

        if check:
            helpers._check_command_response(
                result, self.connection._disconnect, None, allowable_errors)

        return result, cursor.conn_id

#.........这里部分代码省略.........
          - `allowable_errors`: if `check` is ``True``, error messages
            in this list will be ignored by error-checking
          - `uuid_subtype` (optional): The BSON binary subtype to use
            for a UUID used in this command.
          - `compile_re` (optional): if ``False``, don't attempt to compile
            BSON regular expressions into Python regular expressions. Return
            instances of :class:`~bson.regex.Regex` instead. Can avoid
            :exc:`~bson.errors.InvalidBSON` errors when receiving
            Python-incompatible regular expressions, for example from
            ``currentOp``
          - `read_preference`: The read preference for this connection.
            See :class:`~pymongo.read_preferences.ReadPreference` for available
            options.
          - `tag_sets`: Read from replica-set members with these tags.
            To specify a priority-order for tag sets, provide a list of
            tag sets: ``[{'dc': 'ny'}, {'dc': 'la'}, {}]``. A final, empty tag
            set, ``{}``, means "read from any member that matches the mode,
            ignoring tags." ReplicaSetConnection tries each set of tags in turn
            until it finds a set of tags with at least one matching member.
          - `secondary_acceptable_latency_ms`: Any replica-set member whose
            ping time is within secondary_acceptable_latency_ms of the nearest
            member may accept reads. Default 15 milliseconds.
            **Ignored by mongos** and must be configured on the command line.
            See the localThreshold_ option for more information.
          - `**kwargs` (optional): additional keyword arguments will
            be added to the command document before it is sent

        .. note:: ``command`` ignores the ``network_timeout`` parameter.

        .. versionchanged:: 2.7
           Added ``compile_re`` option.
        .. versionchanged:: 2.3
           Added `tag_sets` and `secondary_acceptable_latency_ms` options.
        .. versionchanged:: 2.2
           Added support for `as_class` - the class you want to use for
           the resulting documents
        .. versionchanged:: 1.6
           Added the `value` argument for string commands, and keyword
           arguments for additional command options.
        .. versionchanged:: 1.5
           `command` can be a string in addition to a full document.
        .. versionadded:: 1.4

        .. mongodoc:: commands
        .. _localThreshold: http://docs.mongodb.org/manual/reference/mongos/#cmdoption-mongos--localThreshold
        """

        if isinstance(command, basestring):
            command = SON([(command, value)])

        command_name = command.keys()[0].lower()
        must_use_master = kwargs.pop('_use_master', False)
        if command_name not in rp.secondary_ok_commands:
            must_use_master = True

        # Special-case: mapreduce can go to secondaries only if inline
        if command_name == 'mapreduce':
            out = command.get('out') or kwargs.get('out')
            if not isinstance(out, dict) or not out.get('inline'):
                must_use_master = True

        # Special-case: aggregate with $out cannot go to secondaries.
        if command_name == 'aggregate':
            for stage in kwargs.get('pipeline', []):
                if '$out' in stage:
                    must_use_master = True
                    break

        extra_opts = {
            'as_class': kwargs.pop('as_class', None),
            'slave_okay': kwargs.pop('slave_okay', self.slave_okay),
            '_must_use_master': must_use_master,
            '_uuid_subtype': uuid_subtype
        }

        extra_opts['read_preference'] = kwargs.pop(
            'read_preference',
            self.read_preference)
        extra_opts['tag_sets'] = kwargs.pop(
            'tag_sets',
            self.tag_sets)
        extra_opts['secondary_acceptable_latency_ms'] = kwargs.pop(
            'secondary_acceptable_latency_ms',
            self.secondary_acceptable_latency_ms)
        extra_opts['compile_re'] = compile_re

        fields = kwargs.get('fields')
        if fields is not None and not isinstance(fields, dict):
            kwargs['fields'] = helpers._fields_list_to_dict(fields)

        command.update(kwargs)

        result = self["$cmd"].find_one(command, **extra_opts)

        if check:
            msg = "command %s failed: %%s" % repr(command).replace("%", "%%")
            helpers._check_command_response(result, self.connection.disconnect,
                                            msg, allowable_errors)

        return result

    def __init__(self, collection, spec=None, fields=None, skip=0, limit=0,
                 timeout=True, snapshot=False, tailable=False, sort=None,
                 max_scan=None, as_class=None, slave_okay=False,
                 await_data=False, partial=False, manipulate=True,
                 read_preference=ReadPreference.PRIMARY,
                 tag_sets=[{}], secondary_acceptable_latency_ms=None,
                 exhaust=False, compile_re=True, oplog_replay=False,
                 modifiers=None, _must_use_master=False, _codec_options=None,
                 **kwargs):
        """Create a new cursor.

        Should not be called directly by application developers - see
        :meth:`~pymongo.collection.Collection.find` instead.

        .. mongodoc:: cursors
        """

        # Backport aliases.
        if 'filter' in kwargs:
            spec = kwargs['filter']
        if 'projection' in kwargs:
            fields = kwargs['projection']
        if 'no_cursor_timeout' in kwargs:
            timeout = not kwargs['no_cursor_timeout']
        if 'allow_partial_results' in kwargs:
            partial = kwargs['allow_partial_results']

        if 'cursor_type' in kwargs:
            crt = kwargs['cursor_type']
            if crt not in (CursorType.NON_TAILABLE, CursorType.TAILABLE,
                          CursorType.TAILABLE_AWAIT, CursorType.EXHAUST):
                raise ValueError("not a valid value for cursor_type")
            exhaust = crt == CursorType.EXHAUST
            tailable = crt == CursorType.TAILABLE
            if crt == CursorType.TAILABLE_AWAIT:
                await_data = True
                tailable = True

        if modifiers is not None:
            if not isinstance(modifiers, dict):
                raise TypeError("%s must be an instance of dict or subclass"
                                % (modifiers,))
            if '$snapshot' in modifiers:
                snapshot = modifiers['$snapshot']
            if '$maxScan' in modifiers:
                max_scan = modifiers['$maxScan']

        self.__id = None

        if spec is None:
            spec = {}

        if not isinstance(spec, dict):
            raise TypeError("spec must be an instance of dict")
        if not isinstance(skip, int):
            raise TypeError("skip must be an instance of int")
        if not isinstance(limit, int):
            raise TypeError("limit must be an instance of int")
        if not isinstance(timeout, bool):
            raise TypeError("timeout must be an instance of bool")
        if not isinstance(snapshot, bool):
            raise TypeError("snapshot must be an instance of bool")
        if not isinstance(tailable, bool):
            raise TypeError("tailable must be an instance of bool")
        if not isinstance(slave_okay, bool):
            raise TypeError("slave_okay must be an instance of bool")
        if not isinstance(await_data, bool):
            raise TypeError("await_data must be an instance of bool")
        if not isinstance(partial, bool):
            raise TypeError("partial must be an instance of bool")
        if not isinstance(exhaust, bool):
            raise TypeError("exhaust must be an instance of bool")
        if not isinstance(oplog_replay, bool):
            raise TypeError("oplog_replay must be an instance of bool")

        if fields is not None:
            if not fields:
                fields = {"_id": 1}
            if not isinstance(fields, dict):
                fields = helpers._fields_list_to_dict(fields)

        self.__collection = collection
        self.__spec = spec
        self.__fields = fields
        self.__skip = skip
        self.__limit = limit
        self.__max_time_ms = None
        self.__batch_size = 0
        self.__max = None
        self.__min = None
        self.__modifiers = modifiers and modifiers.copy() or {}


        # Exhaust cursor support
        if self.__collection.database.connection.is_mongos and exhaust:
            raise InvalidOperation('Exhaust cursors are '
                                   'not supported by mongos')
        if limit and exhaust:
            raise InvalidOperation("Can't use limit and exhaust together.")
        self.__exhaust = exhaust
#.........这里部分代码省略.........

    def __init__(self, collection, spec=None, fields=None, skip=0, limit=0,
                 timeout=True, snapshot=False, tailable=False, sort=None,
                 max_scan=None, as_class=None,
                 await_data=False, partial=False, manipulate=True,
                 read_preference=None, exhaust=False, compile_re=True,
                 _uuid_subtype=None):
        """Create a new cursor.

        Should not be called directly by application developers - see
        :meth:`~pymongo.collection.Collection.find` instead.

        .. mongodoc:: cursors
        """
        self.__id = None

        if spec is None:
            spec = {}

        if not isinstance(spec, dict):
            raise TypeError("spec must be an instance of dict")
        if not isinstance(skip, int):
            raise TypeError("skip must be an instance of int")
        if not isinstance(limit, int)