docs: 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,
              parse_write_concern_error=False,
              collation=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
        - `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.
        - `parse_write_concern_error`: Whether to parse the ``writeConcernError``
          field in the command response.
        - `collation`: The collation 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
      if collation is not None:
          spec['collation'] = collation
  
      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,
                  parse_write_concern_error=parse_write_concern_error)
      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