def command(sock, dbname, spec, slave_ok, is_mongos, read_preference,
            codec_options, check=True, allowable_errors=None):
    """Execute a command over the socket, or raise socket.error.

    :Parameters:
      - `sock`: a raw socket instance
      - `dbname`: name of the database on which to run the command
      - `spec`: a command document as a dict, SON, or mapping object
      - `slave_ok`: whether to set the SlaveOkay wire protocol bit
      - `is_mongos`: are we connected to a mongos?
      - `read_preference`: a read preference
      - `codec_options`: a CodecOptions instance
      - `check`: raise OperationFailure if there are errors
      - `allowable_errors`: errors to ignore if `check` is True
    """
    ns = dbname + '.$cmd'
    flags = 4 if slave_ok else 0
    if is_mongos:
        spec = message._maybe_add_read_preference(spec, read_preference)
    request_id, msg, _ = message.query(flags, ns, 0, -1, spec,
                                       None, codec_options)
    sock.sendall(msg)
    response = receive_message(sock, 1, request_id)
    unpacked = helpers._unpack_response(response, codec_options=codec_options)
    response_doc = unpacked['data'][0]
    msg = "command %s on namespace %s failed: %%s" % (
        repr(spec).replace("%", "%%"), ns)
    if check:
        helpers._check_command_response(response_doc, msg, allowable_errors)
    return response_doc

    def __check_response_to_last_error(self, response):
        """Check a response to a lastError message for errors.

        `response` is a byte string representing a response to the message.
        If it represents an error response we raise OperationFailure.

        Return the response as a document.
        """
        response = helpers._unpack_response(response)

        assert response["number_returned"] == 1
        error = response["data"][0]

        helpers._check_command_response(error, self.disconnect)

        # TODO unify logic with database.error method
        if error.get("err", 0) is None:
            return error
        if error["err"] == "not master":
            self.disconnect()
            raise AutoReconnect("not master")

        if "code" in error:
            if error["code"] in [11000, 11001]:
                raise DuplicateKeyError(error["err"])
            else:
                raise OperationFailure(error["err"], error["code"])
        else:
            raise OperationFailure(error["err"])

    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 test_command_response_without_ok(self):
        # Sometimes (SERVER-10891) the server's response to a badly-formatted
        # command document will have no 'ok' field. We should raise
        # OperationFailure instead of KeyError.
        self.assertRaises(OperationFailure, helpers._check_command_response, {}, reset=None)

        try:
            helpers._check_command_response({"$err": "foo"}, reset=None)
        except OperationFailure, e:
            self.assertEqual(e.args[0], "foo")

    def test_mongos_response(self):
        error_document = {
            'ok': 0,
            'errmsg': 'outer',
            'raw': {'shard0/host0,host1': {'ok': 0, 'errmsg': 'inner'}}}

        try:
            helpers._check_command_response(error_document, reset=None)
        except OperationFailure, exc:
            self.assertEqual('inner', str(exc))

 def __simple_command(self, sock, dbname, spec):
     """Send a command to the server.
     """
     rqst_id, msg, _ = message.query(0, dbname + '.$cmd', 0, -1, spec)
     sock.sendall(msg)
     response = self.__recv_msg(1, rqst_id, sock)
     response = helpers._unpack_response(response)['data'][0]
     msg = "command %r failed: %%s" % spec
     helpers._check_command_response(response, None, msg)
     return response

 def __simple_command(self, sock_info, dbname, spec):
     """Send a command to the server.
     """
     rqst_id, msg, _ = message.query(0, dbname + ".$cmd", 0, -1, spec)
     sock_info.sock.sendall(msg)
     response = self.__receive_message_on_socket(1, rqst_id, sock_info)
     response = helpers._unpack_response(response)["data"][0]
     msg = "command %r failed: %%s" % spec
     helpers._check_command_response(response, None, msg)
     return response

    def test_command_response_without_ok(self):
        # Sometimes (SERVER-10891) the server's response to a badly-formatted
        # command document will have no 'ok' field. We should raise
        # OperationFailure instead of KeyError.
        self.assertRaises(OperationFailure,
                          helpers._check_command_response, {})

        try:
            helpers._check_command_response({'$err': 'foo'})
        except OperationFailure as e:
            self.assertEqual(e.args[0], 'foo')
        else:
            self.fail("_check_command_response didn't raise OperationFailure")

    def command(self, command, value=1, check=True, allowable_errors=None, **kwargs):
        if isinstance(command, basestring):
            command = SON([(command, value)])
        command.update(kwargs)

        ns = self["$cmd"]
        response = yield ns.find_one(command)

        if check:
            msg = "command {0} on namespace {1} failed: %s".format(repr(command), ns)
            _check_command_response(response, msg, allowable_errors)

        defer.returnValue(response)

    def command(self, command, value=1, check=True, allowable_errors=None, **kwargs):
        if isinstance(command, (bytes, unicode)):
            command = SON([(command, value)])
        options = kwargs.copy()
        options.pop("_deadline", None)
        command.update(options)

        ns = self["$cmd"]
        response = yield ns.find_one(command, **kwargs)

        if check:
            msg = "TxMongo: command {0} on namespace {1} failed with '%s'".format(repr(command), ns)
            _check_command_response(response, msg, allowable_errors)

        defer.returnValue(response)

    def write_command(self, request_id, msg):
        """Send "insert" etc. command, returning response as a dict.

        Can raise ConnectionFailure or OperationFailure.

        :Parameters:
          - `request_id`: an int.
          - `msg`: bytes, the command message.
        """
        self.send_message(msg, 0)
        reply = self.receive_message(request_id)
        result = reply.command_response()

        # Raises NotMasterError or OperationFailure.
        helpers._check_command_response(result)
        return result

    def write_command(self, request_id, msg):
        """Send "insert" etc. command, returning response as a dict.

        Can raise ConnectionFailure or OperationFailure.

        :Parameters:
          - `request_id`: an int.
          - `msg`: bytes, the command message.
        """
        self.send_message(msg, 0)
        response = helpers._unpack_response(self.receive_message(1, request_id))
        assert response['number_returned'] == 1
        result = response['data'][0]

        # Raises NotMasterError or OperationFailure.
        helpers._check_command_response(result)
        return result

    def __simple_command(self, sock_info, dbname, spec):
        """Send a command to the server.
        """
        rqst_id, msg, _ = message.query(0, dbname + '.$cmd', 0, -1, spec)
        start = time.time()
        try:
            sock_info.sock.sendall(msg)
            response = self.__receive_message_on_socket(1, rqst_id, sock_info)
        except:
            sock_info.close()
            raise

        end = time.time()
        response = helpers._unpack_response(response)['data'][0]
        msg = "command %r failed: %%s" % spec
        helpers._check_command_response(response, None, msg)
        return response, end - start

    def test_mongos_response(self):
        error_document = {
            'ok': 0,
            'errmsg': 'outer',
            'raw': {'shard0/host0,host1': {'ok': 0, 'errmsg': 'inner'}}}

        with self.assertRaises(OperationFailure) as context:
            helpers._check_command_response(error_document)

        self.assertEqual('inner', str(context.exception))

        # If a shard has no primary and you run a command like dbstats, which
        # cannot be run on a secondary, mongos's response includes empty "raw"
        # errors. See SERVER-15428.
        error_document = {
            'ok': 0,
            'errmsg': 'outer',
            'raw': {'shard0/host0,host1': {}}}

        with self.assertRaises(OperationFailure) as context:
            helpers._check_command_response(error_document)

        self.assertEqual('outer', str(context.exception))

        # Raw error has ok: 0 but no errmsg. Not a known case, but test it.
        error_document = {
            'ok': 0,
            'errmsg': 'outer',
            'raw': {'shard0/host0,host1': {'ok': 0}}}

        with self.assertRaises(OperationFailure) as context:
            helpers._check_command_response(error_document)

        self.assertEqual('outer', str(context.exception))

    def __check_response_to_last_error(self, response):
        """Check a response to a lastError message for errors.

        `response` is a byte string representing a response to the message.
        If it represents an error response we raise OperationFailure.

        Return the response as a document.
        """
        response = helpers._unpack_response(response)

        assert response["number_returned"] == 1
        error = response["data"][0]

        helpers._check_command_response(error, self.disconnect)

        error_msg = error.get("err", "")
        if error_msg is None:
            return error
        if error_msg.startswith("not master"):
            self.disconnect()
            raise AutoReconnect(error_msg)

        details = error
        # mongos returns the error code in an error object
        # for some errors.
        if "errObjects" in error:
            for errobj in error["errObjects"]:
                if errobj["err"] == error_msg:
                    details = errobj
                    break

        if "code" in details:
            if details["code"] in (11000, 11001, 12582):
                raise DuplicateKeyError(details["err"], details["code"])
            else:
                raise OperationFailure(details["err"], details["code"])
        else:
            raise OperationFailure(details["err"])

    def __send_message(self, operation):
        """Send a query or getmore operation and handles the response.

        If operation is ``None`` this is an exhaust cursor, which reads
        the next result batch off the exhaust socket instead of
        sending getMore messages to the server.

        Can raise ConnectionFailure.
        """
        client = self.__collection.database.client
        listeners = client._event_listeners
        publish = listeners.enabled_for_commands
        from_command = False
        start = datetime.datetime.now()

        def duration(): return datetime.datetime.now() - start

        if operation:
            kwargs = {
                "read_preference": self.__read_preference,
                "exhaust": self.__exhaust,
            }
            if self.__address is not None:
                kwargs["address"] = self.__address

            try:
                response = client._send_message_with_response(operation,
                                                              **kwargs)
                self.__address = response.address
                if self.__exhaust:
                    # 'response' is an ExhaustResponse.
                    self.__exhaust_mgr = _SocketManager(response.socket_info,
                                                        response.pool)

                cmd_name = operation.name
                reply = response.data
                rqst_id = response.request_id
                from_command = response.from_command
            except AutoReconnect:
                # Don't try to send kill cursors on another socket
                # or to another server. It can cause a _pinValue
                # assertion on some server releases if we get here
                # due to a socket timeout.
                self.__killed = True
                raise
        else:
            # Exhaust cursor - no getMore message.
            rqst_id = 0
            cmd_name = 'getMore'
            if publish:
                # Fake a getMore command.
                cmd = SON([('getMore', self.__id),
                           ('collection', self.__collection.name)])
                if self.__batch_size:
                    cmd['batchSize'] = self.__batch_size
                if self.__max_time_ms:
                    cmd['maxTimeMS'] = self.__max_time_ms
                listeners.publish_command_start(
                    cmd, self.__collection.database.name, 0, self.__address)
            try:
                reply = self.__exhaust_mgr.sock.receive_message(None)
            except Exception as exc:
                if publish:
                    listeners.publish_command_failure(
                        duration(), _convert_exception(exc), cmd_name, rqst_id,
                        self.__address)
                if isinstance(exc, ConnectionFailure):
                    self.__die()
                raise

        try:
            docs = self._unpack_response(response=reply,
                                         cursor_id=self.__id,
                                         codec_options=self.__codec_options)
            if from_command:
                first = docs[0]
                client._receive_cluster_time(first, self.__session)
                helpers._check_command_response(first)
        except OperationFailure as exc:
            self.__killed = True

            # Make sure exhaust socket is returned immediately, if necessary.
            self.__die()

            if publish:
                listeners.publish_command_failure(
                    duration(), exc.details, cmd_name, rqst_id, self.__address)

            # If this is a tailable cursor the error is likely
            # due to capped collection roll over. Setting
            # self.__killed to True ensures Cursor.alive will be
            # False. No need to re-raise.
            if self.__query_flags & _QUERY_OPTIONS["tailable_cursor"]:
                return
            raise
        except NotMasterError as exc:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True

#.........这里部分代码省略.........

#.........这里部分代码省略.........
        >>> 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 __send_message(self, operation):
        """Send a getmore message and handle the response.
        """
        client = self.__collection.database.client
        listeners = client._event_listeners
        publish = listeners.enabled_for_commands
        try:
            response = client._send_message_with_response(
                operation, address=self.__address)
        except AutoReconnect:
            # Don't try to send kill cursors on another socket
            # or to another server. It can cause a _pinValue
            # assertion on some server releases if we get here
            # due to a socket timeout.
            self.__killed = True
            raise

        cmd_duration = response.duration
        rqst_id = response.request_id
        from_command = response.from_command

        if publish:
            start = datetime.datetime.now()
        try:
            doc = helpers._unpack_response(response.data,
                                           self.__id,
                                           self.__collection.codec_options)
            if from_command:
                helpers._check_command_response(doc['data'][0])

        except OperationFailure as exc:
            self.__killed = True

            if publish:
                duration = (datetime.datetime.now() - start) + cmd_duration
                listeners.publish_command_failure(
                    duration, exc.details, "getMore", rqst_id, self.__address)

            raise
        except NotMasterError as exc:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            self.__killed = True

            if publish:
                duration = (datetime.datetime.now() - start) + cmd_duration
                listeners.publish_command_failure(
                    duration, exc.details, "getMore", rqst_id, self.__address)

            client._reset_server_and_request_check(self.address)
            raise
        except Exception as exc:
            if publish:
                duration = (datetime.datetime.now() - start) + cmd_duration
                listeners.publish_command_failure(
                    duration, _convert_exception(exc), "getMore", rqst_id,
                    self.__address)
            raise

        if from_command:
            cursor = doc['data'][0]['cursor']
            documents = cursor['nextBatch']
            self.__id = cursor['id']
            self.__retrieved += len(documents)
        else:
            documents = doc["data"]
            self.__id = doc["cursor_id"]
            self.__retrieved += doc["number_returned"]

        if publish:
            duration = (datetime.datetime.now() - start) + cmd_duration
            # Must publish in getMore command response format.
            res = {"cursor": {"id": self.__id,
                              "ns": self.__collection.full_name,
                              "nextBatch": documents},
                   "ok": 1}
            listeners.publish_command_success(
                duration, res, "getMore", rqst_id, self.__address)

        if self.__id == 0:
            self.__killed = True
        self.__data = deque(documents)

    def __send_message(self, operation):
        """Send a getmore message and handle the response.
        """
        def kill():
            self.__killed = True
            self.__end_session(True)

        client = self.__collection.database.client
        listeners = client._event_listeners
        publish = listeners.enabled_for_commands
        start = datetime.datetime.now()

        def duration(): return datetime.datetime.now() - start

        try:
            response = client._send_message_with_response(
                operation, address=self.__address)
        except AutoReconnect:
            # Don't try to send kill cursors on another socket
            # or to another server. It can cause a _pinValue
            # assertion on some server releases if we get here
            # due to a socket timeout.
            kill()
            raise

        rqst_id = response.request_id
        from_command = response.from_command
        reply = response.data

        try:
            docs = self._unpack_response(reply,
                                         self.__id,
                                         self.__collection.codec_options)
            if from_command:
                first = docs[0]
                client._receive_cluster_time(first, self.__session)
                helpers._check_command_response(first)

        except OperationFailure as exc:
            kill()

            if publish:
                listeners.publish_command_failure(
                    duration(), exc.details, "getMore", rqst_id, self.__address)

            raise
        except NotMasterError as exc:
            # Don't send kill cursors to another server after a "not master"
            # error. It's completely pointless.
            kill()

            if publish:
                listeners.publish_command_failure(
                    duration(), exc.details, "getMore", rqst_id, self.__address)

            client._reset_server_and_request_check(self.address)
            raise
        except Exception as exc:
            if publish:
                listeners.publish_command_failure(
                    duration(), _convert_exception(exc), "getMore", rqst_id,
                    self.__address)
            raise

        if from_command:
            cursor = docs[0]['cursor']
            documents = cursor['nextBatch']
            self.__id = cursor['id']
            if publish:
                listeners.publish_command_success(
                    duration(), docs[0], "getMore", rqst_id,
                    self.__address)
        else:
            documents = docs
            self.__id = reply.cursor_id

            if publish:
                # Must publish in getMore command response format.
                res = {"cursor": {"id": self.__id,
                                  "ns": self.__collection.full_name,
                                  "nextBatch": documents},
                       "ok": 1}
                listeners.publish_command_success(
                    duration(), res, "getMore", rqst_id, self.__address)

        if self.__id == 0:
            kill()
        self.__data = deque(documents)

def command(sock, dbname, spec, slave_ok, is_mongos,
            read_preference, codec_options, check=True,
            allowable_errors=None, address=None,
            check_keys=False, listeners=None, max_bson_size=None,
            read_concern=DEFAULT_READ_CONCERN):
    """Execute a command over the socket, or raise socket.error.

    :Parameters:
      - `sock`: a raw socket instance
      - `dbname`: name of the database on which to run the command
      - `spec`: a command document as a dict, SON, or mapping object
      - `slave_ok`: whether to set the SlaveOkay wire protocol bit
      - `is_mongos`: are we connected to a mongos?
      - `read_preference`: a read preference
      - `codec_options`: a CodecOptions instance
      - `check`: raise OperationFailure if there are errors
      - `allowable_errors`: errors to ignore if `check` is True
      - `address`: the (host, port) of `sock`
      - `check_keys`: if True, check `spec` for invalid keys
      - `listeners`: An instance of :class:`~pymongo.monitoring.EventListeners`
      - `max_bson_size`: The maximum encoded bson size for this server
      - `read_concern`: The read concern for this command.
    """
    name = next(iter(spec))
    ns = dbname + '.$cmd'
    flags = 4 if slave_ok else 0
    # Publish the original command document.
    orig = spec
    if is_mongos:
        spec = message._maybe_add_read_preference(spec, read_preference)
    if read_concern.level:
        spec['readConcern'] = read_concern.document

    publish = listeners is not None and listeners.enabled_for_commands
    if publish:
        start = datetime.datetime.now()

    request_id, msg, size = message.query(flags, ns, 0, -1, spec,
                                          None, codec_options, check_keys)

    if (max_bson_size is not None
            and size > max_bson_size + message._COMMAND_OVERHEAD):
        message._raise_document_too_large(
            name, size, max_bson_size + message._COMMAND_OVERHEAD)

    if publish:
        encoding_duration = datetime.datetime.now() - start
        listeners.publish_command_start(orig, dbname, request_id, address)
        start = datetime.datetime.now()

    try:
        sock.sendall(msg)
        response = receive_message(sock, 1, request_id)
        unpacked = helpers._unpack_response(
            response, codec_options=codec_options)

        response_doc = unpacked['data'][0]
        if check:
            helpers._check_command_response(response_doc, None, allowable_errors)
    except Exception as exc:
        if publish:
            duration = (datetime.datetime.now() - start) + encoding_duration
            if isinstance(exc, (NotMasterError, OperationFailure)):
                failure = exc.details
            else:
                failure = message._convert_exception(exc)
            listeners.publish_command_failure(
                duration, failure, name, request_id, address)
        raise
    if publish:
        duration = (datetime.datetime.now() - start) + encoding_duration
        listeners.publish_command_success(
            duration, response_doc, name, request_id, address)
    return response_doc