Python pymongo.errors 模块，OperationFailure() 实例源码

def _refresh(self):
        """Refreshes the cursor with more data from the server.

        Returns the length of self.__data after refresh. Will exit early if
        self.__data is already non-empty. Raises OperationFailure when the
        cursor cannot be refreshed due to an error on the query.
        """
        if len(self.__data) or self.__killed:
            return len(self.__data)

        if self.__id:  # Get More
            dbname, collname = self.__ns.split('.', 1)
            self.__send_message(
                _GetMore(dbname,
                         collname,
                         self.__batch_size,
                         self.__id,
                         self.__collection.codec_options))

        else:  # Cursor id is zero nothing else to return
            self.__killed = True

        return len(self.__data)

def __create_index(self, keys, index_options):
        """Internal create index helper.

        :Parameters:
          - `keys`: a list of tuples [(key, type), (key, type), ...]
          - `index_options`: a dict of index options.
        """
        index_doc = helpers._index_document(keys)
        index = {"key": index_doc}
        index.update(index_options)

        with self._socket_for_writes() as sock_info:
            cmd = SON([('createIndexes', self.name), ('indexes', [index])])
            try:
                self._command(
                    sock_info, cmd, read_preference=ReadPreference.PRIMARY)
            except OperationFailure as exc:
                if exc.code in common.COMMAND_NOT_FOUND_CODES:
                    index["ns"] = self.__full_name
                    wcn = (self.write_concern if
                           self.write_concern.acknowledged else WriteConcern())
                    self.__database.system.indexes._insert(
                        sock_info, index, True, False, False, wcn)
                else:
                    raise

def write_command(self, request_id, msg, docs):
        """A proxy for SocketInfo.write_command that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            reply = self.sock_info.write_command(request_id, msg)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(request_id, exc.details, duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return reply

def connect(self):
        try:
            logging.debug("Getting MongoDB connection to %s (replicaSet=%s, readPreference=%s, readPreferenceTags=%s, ssl=%s)" % (
                self.uri,
                self.replset,
                self.read_pref,
                self.do_rp_tags,
                self.do_ssl(),
            ))
            conn = MongoClient(**self.client_opts())
            if self.do_connect:
                conn['admin'].command({"ping": 1})
        except (ConnectionFailure, OperationFailure, ServerSelectionTimeoutError), e:
            logging.error("Unable to connect to %s! Error: %s" % (self.uri, e))
            raise DBConnectionError(e)
        if conn is not None:
            self._conn = conn
        return self._conn

def test_validate_collection(self, test_db):
        with pytest.raises(TypeError):
            await test_db.validate_collection(5)
        with pytest.raises(TypeError):
            await test_db.validate_collection(None)

        await test_db.test.insert_one({'dummy': 'object'})

        with pytest.raises(OperationFailure):
            await test_db.validate_collection('test.doesnotexist"')
        with pytest.raises(OperationFailure):
            await test_db.validate_collection(test_db.test.doesnotexist)

        assert await test_db.validate_collection('test')
        assert await test_db.validate_collection(test_db.test)
        assert await test_db.validate_collection(test_db.test, full=True)
        assert await test_db.validate_collection(test_db.test, scandata=True)
        assert await test_db.validate_collection(test_db.test, full=True, scandata=True)
        assert await test_db.validate_collection(test_db.test, True, True)

def _refresh(self):
        """Refreshes the cursor with more data from the server.

        Returns the length of self.__data after refresh. Will exit early if
        self.__data is already non-empty. Raises OperationFailure when the
        cursor cannot be refreshed due to an error on the query.
        """
        if len(self.__data) or self.__killed:
            return len(self.__data)

        if self.__id:  # Get More
            dbname, collname = self.__ns.split('.', 1)
            self.__send_message(
                _GetMore(dbname,
                         collname,
                         self.__batch_size,
                         self.__id,
                         self.__collection.codec_options))

        else:  # Cursor id is zero nothing else to return
            self.__killed = True

        return len(self.__data)

def __create_index(self, keys, index_options):
        """Internal create index helper.

        :Parameters:
          - `keys`: a list of tuples [(key, type), (key, type), ...]
          - `index_options`: a dict of index options.
        """
        index_doc = helpers._index_document(keys)
        index = {"key": index_doc}
        index.update(index_options)

        with self._socket_for_writes() as sock_info:
            cmd = SON([('createIndexes', self.name), ('indexes', [index])])
            try:
                self._command(
                    sock_info, cmd, read_preference=ReadPreference.PRIMARY)
            except OperationFailure as exc:
                if exc.code in common.COMMAND_NOT_FOUND_CODES:
                    index["ns"] = self.__full_name
                    wcn = (self.write_concern if
                           self.write_concern.acknowledged else WriteConcern())
                    self.__database.system.indexes._insert(
                        sock_info, index, True, False, False, wcn)
                else:
                    raise

def write_command(self, request_id, msg, docs):
        """A proxy for SocketInfo.write_command that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            reply = self.sock_info.write_command(request_id, msg)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(request_id, exc.details, duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return reply

def _refresh(self):
        """Refreshes the cursor with more data from the server.

        Returns the length of self.__data after refresh. Will exit early if
        self.__data is already non-empty. Raises OperationFailure when the
        cursor cannot be refreshed due to an error on the query.
        """
        if len(self.__data) or self.__killed:
            return len(self.__data)

        if self.__id:  # Get More
            dbname, collname = self.__ns.split('.', 1)
            self.__send_message(
                _GetMore(dbname,
                         collname,
                         self.__batch_size,
                         self.__id,
                         self.__collection.codec_options))

        else:  # Cursor id is zero nothing else to return
            self.__killed = True

        return len(self.__data)

def __create_index(self, keys, index_options):
        """Internal create index helper.

        :Parameters:
          - `keys`: a list of tuples [(key, type), (key, type), ...]
          - `index_options`: a dict of index options.
        """
        index_doc = helpers._index_document(keys)
        index = {"key": index_doc}
        index.update(index_options)

        with self._socket_for_writes() as sock_info:
            cmd = SON([('createIndexes', self.name), ('indexes', [index])])
            try:
                self._command(
                    sock_info, cmd, read_preference=ReadPreference.PRIMARY)
            except OperationFailure as exc:
                if exc.code in common.COMMAND_NOT_FOUND_CODES:
                    index["ns"] = self.__full_name
                    wcn = (self.write_concern if
                           self.write_concern.acknowledged else WriteConcern())
                    self.__database.system.indexes._insert(
                        sock_info, index, True, False, False, wcn)
                else:
                    raise

def write_command(self, request_id, msg, docs):
        """A proxy for SocketInfo.write_command that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            reply = self.sock_info.write_command(request_id, msg)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(request_id, exc.details, duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return reply

def _refresh(self):
        """Refreshes the cursor with more data from the server.

        Returns the length of self.__data after refresh. Will exit early if
        self.__data is already non-empty. Raises OperationFailure when the
        cursor cannot be refreshed due to an error on the query.
        """
        if len(self.__data) or self.__killed:
            return len(self.__data)

        if self.__id:  # Get More
            dbname, collname = self.__ns.split('.', 1)
            self.__send_message(
                _GetMore(dbname,
                         collname,
                         self.__batch_size,
                         self.__id,
                         self.__collection.codec_options))

        else:  # Cursor id is zero nothing else to return
            self.__killed = True

        return len(self.__data)

def drop_indexes(self):
        """Drops all indexes on this collection.

        Can be used on non-existant collections or collections with no indexes.
        Raises OperationFailure on an error.

        .. note:: The :attr:`~pymongo.collection.Collection.write_concern` of
           this collection is automatically applied to this operation when using
           MongoDB >= 3.4.

        .. versionchanged:: 3.4
           Apply this collection's write concern automatically to this operation
           when connected to MongoDB >= 3.4.

        """
        self.__database.client._purge_index(self.__database.name, self.__name)
        self.drop_index("*")

def write_command(self, request_id, msg, docs):
        """A proxy for SocketInfo.write_command that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            reply = self.sock_info.write_command(request_id, msg)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(request_id, exc.details, duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return reply

def _refresh(self):
        """Refreshes the cursor with more data from the server.

        Returns the length of self.__data after refresh. Will exit early if
        self.__data is already non-empty. Raises OperationFailure when the
        cursor cannot be refreshed due to an error on the query.
        """
        if len(self.__data) or self.__killed:
            return len(self.__data)

        if self.__id:  # Get More
            dbname, collname = self.__ns.split('.', 1)
            self.__send_message(
                _GetMore(dbname,
                         collname,
                         self.__batch_size,
                         self.__id,
                         self.__collection.codec_options))

        else:  # Cursor id is zero nothing else to return
            self.__killed = True

        return len(self.__data)

def __create_index(self, keys, index_options):
        """Internal create index helper.

        :Parameters:
          - `keys`: a list of tuples [(key, type), (key, type), ...]
          - `index_options`: a dict of index options.
        """
        index_doc = helpers._index_document(keys)
        index = {"key": index_doc}
        index.update(index_options)

        with self._socket_for_writes() as sock_info:
            cmd = SON([('createIndexes', self.name), ('indexes', [index])])
            try:
                self._command(
                    sock_info, cmd, read_preference=ReadPreference.PRIMARY)
            except OperationFailure as exc:
                if exc.code in common.COMMAND_NOT_FOUND_CODES:
                    index["ns"] = self.__full_name
                    wcn = (self.write_concern if
                           self.write_concern.acknowledged else WriteConcern())
                    self.__database.system.indexes._insert(
                        sock_info, index, True, False, False, wcn)
                else:
                    raise

def write_command(self, request_id, msg, docs):
        """A proxy for SocketInfo.write_command that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            reply = self.sock_info.write_command(request_id, msg)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(request_id, exc.details, duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return reply

def _refresh(self):
        """Refreshes the cursor with more data from the server.

        Returns the length of self.__data after refresh. Will exit early if
        self.__data is already non-empty. Raises OperationFailure when the
        cursor cannot be refreshed due to an error on the query.
        """
        if len(self.__data) or self.__killed:
            return len(self.__data)

        if self.__id:  # Get More
            self.__send_message(
                _GetMore(self.__collection.database.name,
                         self.__collection.name,
                         self.__batch_size,
                         self.__id,
                         self.__collection.codec_options))

        else:  # Cursor id is zero nothing else to return
            self.__killed = True

        return len(self.__data)

def __create_index(self, keys, index_options):
        """Internal create index helper.

        :Parameters:
          - `keys`: a list of tuples [(key, type), (key, type), ...]
          - `index_options`: a dict of index options.
        """
        index_doc = helpers._index_document(keys)
        index = {"key": index_doc}
        index.update(index_options)

        with self._socket_for_writes() as sock_info:
            cmd = SON([('createIndexes', self.name), ('indexes', [index])])
            try:
                self._command(
                    sock_info, cmd, read_preference=ReadPreference.PRIMARY)
            except OperationFailure as exc:
                if exc.code in common.COMMAND_NOT_FOUND_CODES:
                    index["ns"] = self.__full_name
                    wcn = (self.write_concern if
                           self.write_concern.acknowledged else WriteConcern())
                    self.__database.system.indexes._insert(
                        sock_info, index, True, False, False, wcn)
                else:
                    raise

def write_command(self, request_id, msg, docs):
        """A proxy for SocketInfo.write_command that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            reply = self.sock_info.write_command(request_id, msg)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(request_id, exc.details, duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return reply

def _refresh(self):
        """Refreshes the cursor with more data from the server.

        Returns the length of self.__data after refresh. Will exit early if
        self.__data is already non-empty. Raises OperationFailure when the
        cursor cannot be refreshed due to an error on the query.
        """
        if len(self.__data) or self.__killed:
            return len(self.__data)

        if self.__id:  # Get More
            self.__send_message(
                _GetMore(self.__collection.database.name,
                         self.__collection.name,
                         self.__batch_size,
                         self.__id,
                         self.__collection.codec_options))

        else:  # Cursor id is zero nothing else to return
            self.__killed = True

        return len(self.__data)

def write_command(self, request_id, msg, docs):
        """A proxy for SocketInfo.write_command that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            reply = self.sock_info.write_command(request_id, msg)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(request_id, exc.details, duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return reply

def update(self, uinfo, user, password, roles):
        if roles_changed(uinfo, roles, self.database):
            self.add(user, password, roles)
        else:
            test_client = self.get_client()

            db = test_client[self.database]

            try:
                db.authenticate(user, password)

                self.module.exit_json(changed=False, user=user)
            except OperationFailure:
                # If we get an operation failure, assume authentication failed, meaning we need to change the password
                # This is...so not good practice, but it's a way to get idempotence from our task
                self.add(user, password, roles)

def _command(self, sock_info, command, slave_ok=False,
                 read_preference=None,
                 codec_options=None, check=True, allowable_errors=None,
                 read_concern=DEFAULT_READ_CONCERN):
        """Internal command helper.

        :Parameters:
          - `sock_info` - A SocketInfo instance.
          - `command` - The command itself, as a SON instance.
          - `slave_ok`: whether to set the SlaveOkay wire protocol bit.
          - `codec_options` (optional) - An instance of
            :class:`~bson.codec_options.CodecOptions`.
          - `check`: raise OperationFailure if there are errors
          - `allowable_errors`: errors to ignore if `check` is True
          - `read_concern` (optional) - An instance of
            :class:`~pymongo.read_concern.ReadConcern`.

        :Returns:

            # todo: don't return address

          (result document, address of server the command was run on)
        """
        return sock_info.command(self.__database.name,
                                 command,
                                 slave_ok,
                                 read_preference or self.read_preference,
                                 codec_options or self.codec_options,
                                 check,
                                 allowable_errors,
                                 read_concern=read_concern)

def drop_indexes(self):
        """Drops all indexes on this collection.

        Can be used on non-existant collections or collections with no indexes.
        Raises OperationFailure on an error.
        """
        self.__database.client._purge_index(self.__database.name, self.__name)
        self.drop_index("*")

def legacy_write(self, request_id, msg, max_doc_size, acknowledged, docs):
        """A proxy for SocketInfo.legacy_write that handles event publishing.
        """
        if self.publish:
            duration = datetime.datetime.now() - self.start_time
            cmd = self._start(request_id, docs)
            start = datetime.datetime.now()
        try:
            result = self.sock_info.legacy_write(
                request_id, msg, max_doc_size, acknowledged)
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                if result is not None:
                    reply = _convert_write_result(self.name, cmd, result)
                else:
                    # Comply with APM spec.
                    reply = {'ok': 1}
                self._succeed(request_id, reply, duration)
        except OperationFailure as exc:
            if self.publish:
                duration = (datetime.datetime.now() - start) + duration
                self._fail(
                    request_id,
                    _convert_write_result(
                        self.name, cmd, exc.details),
                    duration)
            raise
        finally:
            self.start_time = datetime.datetime.now()
        return result

def drop_indexes(self):
        """Drops all indexes on this collection.

        Can be used on non-existant collections or collections with no indexes.
        Raises OperationFailure on an error.
        """
        self.__database.connection._purge_index(self.__database.name,
                                                self.__name)
        self.drop_index(u"*")

def drop_index(self, index_or_name):
        """Drops the specified index on this collection.

        Can be used on non-existant collections or collections with no
        indexes.  Raises OperationFailure on an error. `index_or_name`
        can be either an index name (as returned by `create_index`),
        or an index specifier (as passed to `create_index`). An index
        specifier should be a list of (key, direction) pairs. Raises
        TypeError if index is not an instance of (str, unicode, list).

        .. warning::

          if a custom name was used on index creation (by
          passing the `name` parameter to :meth:`create_index` or
          :meth:`ensure_index`) the index **must** be dropped by name.

        :Parameters:
          - `index_or_name`: index (or name of index) to drop
        """
        name = index_or_name
        if isinstance(index_or_name, list):
            name = _gen_index_name(index_or_name)

        if not isinstance(name, basestring):
            raise TypeError("index_or_name must be an index name or list")

        self.__database.connection._purge_index(self.__database.name,
                                                self.__name, name)
        self.__database.command("dropIndexes", self.__name, index=name,
                                allowable_errors=["ns not found"])

def drop_indexes(self):
        """Drops all indexes on this collection.

        Can be used on non-existant collections or collections with no indexes.
        Raises OperationFailure on an error.
        """
        self.__database.connection._purge_index(self.__database.name,
                                                self.__name)
        self.drop_index(u"*")

def drop_index(self, index_or_name):
        """Drops the specified index on this collection.

        Can be used on non-existant collections or collections with no
        indexes.  Raises OperationFailure on an error. `index_or_name`
        can be either an index name (as returned by `create_index`),
        or an index specifier (as passed to `create_index`). An index
        specifier should be a list of (key, direction) pairs. Raises
        TypeError if index is not an instance of (str, unicode, list).

        .. warning::

          if a custom name was used on index creation (by
          passing the `name` parameter to :meth:`create_index` or
          :meth:`ensure_index`) the index **must** be dropped by name.

        :Parameters:
          - `index_or_name`: index (or name of index) to drop
        """
        name = index_or_name
        if isinstance(index_or_name, list):
            name = _gen_index_name(index_or_name)

        if not isinstance(name, basestring):
            raise TypeError("index_or_name must be an index name or list")

        self.__database.connection._purge_index(self.__database.name,
                                                self.__name, name)
        self.__database.command("dropIndexes", self.__name, index=name,
                                allowable_errors=["ns not found"])

def drop_indexes(self):
        """Drops all indexes on this collection.

        Can be used on non-existant collections or collections with no indexes.
        Raises OperationFailure on an error.
        """
        self.__database.connection._purge_index(self.__database.name,
                                                self.__name)
        self.drop_index(u"*")

def drop_indexes(self):
        """Drops all indexes on this collection.

        Can be used on non-existant collections or collections with no indexes.
        Raises OperationFailure on an error.
        """
        self.__database.connection._purge_index(self.__database.name,
                                                self.__name)
        self.drop_index(u"*")

def drop_index(self, index_or_name):
        """Drops the specified index on this collection.

        Can be used on non-existant collections or collections with no
        indexes.  Raises OperationFailure on an error (e.g. trying to
        drop an index that does not exist). `index_or_name`
        can be either an index name (as returned by `create_index`),
        or an index specifier (as passed to `create_index`). An index
        specifier should be a list of (key, direction) pairs. Raises
        TypeError if index is not an instance of (str, unicode, list).

        .. warning::

          if a custom name was used on index creation (by
          passing the `name` parameter to :meth:`create_index` or
          :meth:`ensure_index`) the index **must** be dropped by name.

        :Parameters:
          - `index_or_name`: index (or name of index) to drop
        """
        name = index_or_name
        if isinstance(index_or_name, list):
            name = _gen_index_name(index_or_name)

        if not isinstance(name, basestring):
            raise TypeError("index_or_name must be an index name or list")

        self.__database.connection._purge_index(self.__database.name,
                                                self.__name, name)
        self.__database.command("dropIndexes", self.__name,
                                read_preference=ReadPreference.PRIMARY,
                                index=name,
                                allowable_errors=["ns not found"])

def auth_if_required(self):
        if self.username is not None and self.password is not None:
            try:
                logging.debug("Authenticating connection with username: %s" % self.username)
                self._conn[self.authdb].authenticate(self.username, self.password)
            except OperationFailure, e:
                logging.fatal("Unable to authenticate with host %s: %s" % (self.uri, e))
                raise DBAuthenticationError(e)
        else:
            pass

def admin_command(self, admin_command, quiet=False):
        tries  = 0
        status = None
        while not status and tries < self.retries:
            try:
                status = self._conn['admin'].command(admin_command)
            except OperationFailure, e:
                if not quiet:
                    logging.error("Error running admin command '%s': %s" % (admin_command, e))
                tries += 1
                sleep(1)
        if not status:
            raise DBOperationError("Could not get output from command: '%s' after %i retries!" % (admin_command, self.retries))
        return status

def is_master(self, force=False):
        try:
            if force or not self._is_master:
                self._is_master = self.admin_command('isMaster', True)
        except OperationFailure, e:
            raise DBOperationError("Unable to run isMaster command! Error: %s" % e)
        return self._is_master

def mark_entry_not_calculated(self, key):
        try:
            self.mongo_collection.update_one(
                filter={
                    'func': _MongoCore._get_func_str(self.func),
                    'key': key
                },
                update={
                    '$set': {'being_calculated': False}
                },
                upsert=False  # should not insert in this case
            )
        except OperationFailure:
            pass  # don't care in this case

def update_one(self, *args, **kwargs):
        raise OperationFailure(Exception())

def test_mongo_write_failure():
    """Testing MongoDB core handling of writing failure scenarios."""
    with pytest.raises(OperationFailure):
        val1 = _func_w_bad_mongo(1, 2)
        val2 = _func_w_bad_mongo(1, 2)
        assert val1 == val2

def test_hint(self, test_db):
        with pytest.raises(TypeError):
            test_db.test.find().hint(3.5)
        await test_db.test.drop()

        await test_db.test.insert_many([{'num': i, 'foo': i} for i in range(100)])

        with pytest.raises(OperationFailure):
            await test_db.test.find({'num': 17, 'foo': 17}).hint([('num', ASCENDING)]).explain()

        with pytest.raises(OperationFailure):
            await test_db.test.find({'num': 17, 'foo': 17}).hint([('foo', ASCENDING)]).explain()

        spec = [('num', DESCENDING)]
        await test_db.test.create_index(spec)

        first = await self._next(test_db.test.find())
        assert 0 == first.get('num')
        first = await self._next(test_db.test.find().hint(spec))
        assert 99 == first.get('num')
        with pytest.raises(OperationFailure):
            await test_db.test.find({'num': 17, 'foo': 17}).hint([('foo', ASCENDING)]).explain()

        a = test_db.test.find({'num': 17})
        a.hint(spec)
        async for _ in a:
            break

        with pytest.raises(InvalidOperation):
            a.hint(spec)

def test_max(self, test_db):
        await test_db.test.create_index([('j', ASCENDING)])

        await test_db.test.insert_many([{'j': j, 'k': j} for j in range(10)])

        cursor = test_db.test.find().max([('j', 3)])
        assert len(await cursor.to_list()) == 3

        # Tuple.
        cursor = test_db.test.find().max((('j', 3),))
        assert len(await cursor.to_list()) == 3

        # Compound index.
        await test_db.test.create_index([('j', ASCENDING), ('k', ASCENDING)])
        cursor = test_db.test.find().max([('j', 3), ('k', 3)])
        assert len(await cursor.to_list()) == 3

        # Wrong order.
        cursor = test_db.test.find().max([('k', 3), ('j', 3)])
        with pytest.raises(OperationFailure):
            await cursor.to_list()

        # No such index.
        cursor = test_db.test.find().max([('k', 3)])
        with pytest.raises(OperationFailure):
            await cursor.to_list()

        with pytest.raises(TypeError):
            test_db.test.find().max(10)

        with pytest.raises(TypeError):
            test_db.test.find().max({'j': 10})

def test_min(self, test_db):
        await test_db.test.create_index([('j', ASCENDING)])

        await test_db.test.insert_many([{'j': j, 'k': j} for j in range(10)])

        cursor = test_db.test.find().min([('j', 3)])
        assert len(await cursor.to_list()) == 7

        # Tuple.
        cursor = test_db.test.find().min((('j', 3),))
        assert len(await cursor.to_list()) == 7

        # Compound index.
        await test_db.test.create_index([('j', ASCENDING), ('k', ASCENDING)])
        cursor = test_db.test.find().min([('j', 3), ('k', 3)])
        assert len(await cursor.to_list()) == 7

        # Wrong order.
        cursor = test_db.test.find().min([('k', 3), ('j', 3)])
        with pytest.raises(OperationFailure):
            await cursor.to_list()

        # No such index.
        cursor = test_db.test.find().min([('k', 3)])
        with pytest.raises(OperationFailure):
            await cursor.to_list()

        with pytest.raises(TypeError):
            test_db.test.find().min(10)

        with pytest.raises(TypeError):
            test_db.test.find().min({'j': 10})

def test_count_with_hint(self, test_db, mongo_version):
        collection = test_db.test

        await collection.insert_many([{'i': 1}, {'i': 2}])
        assert 2 == await collection.find().count()

        await collection.create_index([('i', 1)])

        assert 1 == await collection.find({'i': 1}).hint('_id_').count()
        assert 2 == await collection.find().hint('_id_').count()

        with pytest.raises(OperationFailure):
            await collection.find({'i': 1}).hint('BAD HINT').count()

        # Create a sparse index which should have no entries.
        await collection.create_index([('x', 1)], sparse=True)

        assert 0 == await collection.find({'i': 1}).hint('x_1').count()
        assert 0 == await collection.find({'i': 1}).hint([('x', 1)]).count()

        if mongo_version.at_least(3, 3, 2):
            assert 0 == await collection.find().hint('x_1').count()
            assert 0 == await collection.find().hint([('x', 1)]).count()
        else:
            assert 2 == await collection.find().hint('x_1').count()
            assert 2 == await collection.find().hint([('x', 1)]).count()

def __create_index(self, collection, index, unique):
        doc = await collection.find_one(projection={'_id': 1})
        if doc is None:
            indexes = list()
            try:
                async with await collection.list_indexes() as cursor:
                    async for index in cursor:
                        indexes.append(index)
            except OperationFailure:
                pass
            if index not in indexes:
                await collection.create_index(index, unique=unique)

def drop_index(self, index_or_name: Union[str, list]) -> None:
        """Drops the specified index on this collection.

        Can be used on non-existant collections or collections with no
        indexes.  Raises OperationFailure on an error (e.g. trying to
        drop an index that does not exist). `index_or_name`
        can be either an index name (as returned by `create_index`),
        or an index specifier (as passed to `create_index`). An index
        specifier should be a list of (key, direction) pairs. Raises
        TypeError if index is not an instance of (str, unicode, list).

        .. warning::

          if a custom name was used on index creation (by
          passing the `name` parameter to :meth:`create_index` or
          :meth:`ensure_index`) the index **must** be dropped by name.

        :Parameters:
          - `index_or_name`: index (or name of index) to drop
        """
        name = index_or_name
        if isinstance(index_or_name, list):
            name = helpers._gen_index_name(index_or_name)

        if not isinstance(name, str):
            raise TypeError('index_or_name must be an index name or list')

        cmd = SON([('dropIndexes', self.name), ('index', name)])
        connection = await self.database.client.get_connection()
        await connection.command(
            self.database.name, cmd, ReadPreference.PRIMARY, self.codec_options,
            allowable_errors=['ns not found'], write_concern=self.write_concern,
            parse_write_concern_error=True
        )

def drop_indexes(self) -> None:
        """Drops all indexes on this collection.

        Can be used on non-existant collections or collections with no indexes.
        Raises OperationFailure on an error.
        """
        await self.drop_index('*')

def insert_one(self, document: MutableMapping, bypass_document_validation: bool = False,
                         check_keys: bool = True) -> InsertOneResult:

        common.validate_is_document_type('document', document)

        if '_id' not in document and not isinstance(document, RawBSONDocument):
            document['_id'] = ObjectId()

        write_concern = self.write_concern.document
        acknowledged = write_concern.get('w') != 0

        connection = await self.database.client.get_connection()

        if acknowledged:
            command = SON([('insert', self.name),
                           ('ordered', True),
                           ('documents', [document])])

            if bypass_document_validation and connection.max_wire_version >= 4:
                command['bypassDocumentValidation'] = True

            result = await connection.command(
                self.database.name, command, ReadPreference.PRIMARY, self.__write_response_codec_options,
                check_keys=check_keys
            )

            helpers._check_write_command_response([(0, result)])
        else:
            if bypass_document_validation and connection.max_wire_version >= 4:
                raise OperationFailure('Cannot set bypass_document_validation with',
                                       ' unacknowledged write concern')

            _, msg, _ = message.insert(
                str(self), [document], check_keys,
                acknowledged, write_concern, False, self.__write_response_codec_options
            )
            connection.send_message(msg)

        document_id = document['_id'] if not isinstance(document, RawBSONDocument) else None

        return InsertOneResult(document_id, acknowledged)

def __init__(self, mcfg, col):
        self.tz = mcfg.tz
        self.col = mcfg.db[col]
        try:
            if 'created_at_1' in self.col.index_information():
                if self.col.index_information()['created_at_1'].get('expireAfterSeconds', 0) != mcfg.expire:
                    self.col.drop_index('created_at_1')
                    print('recreated index of "created_at"')
                    self.col.create_index([("created_at", 1)], expireAfterSeconds=mcfg.expire)
            else:
                print('created index of "created_at"')
                self.col.create_index([("created_at", 1)], expireAfterSeconds=mcfg.expire)
        except OperationFailure:
            print('created index of "created_at"')
            self.col.create_index([("created_at", 1)], expireAfterSeconds=mcfg.expire)

def user_add(module, client, db_name, user, password, roles):
    #pymono's user_add is a _create_or_update_user so we won't know if it was changed or updated
    #without reproducing a lot of the logic in database.py of pymongo
    db = client[db_name]
    if roles is None:
        db.add_user(user, password, False)
    else:
        try:
            db.add_user(user, password, None, roles=roles)
        except OperationFailure, e:
            err_msg = str(e)
            module.fail_json(msg=err_msg)

def remove_host(module, client, host_name, timeout=180):
    while True:
        try:
            admin_db = client['admin']
            local_db = client['local']

            if local_db.system.replset.count() > 1:
                module.fail_json(msg='local.system.replset has unexpected contents')

            cfg = local_db.system.replset.find_one()
            if not cfg:
                module.fail_json(msg='no config object retrievable from local.system.replset')

            cfg['version'] += 1

            if len(cfg['members']) == 1:
                module.fail_json(msg="You can't delete last member of replica set")
            for member in cfg['members']:
                if host_name in member['host']:
                    cfg['members'].remove(member)
                else:
                    fail_msg = "couldn't find member with hostname: {0} in replica set members list".format(host_name)
                    module.fail_json(msg=fail_msg)
        except (OperationFailure, AutoReconnect) as e:
            timeout = timeout - 5
            if timeout <= 0:
                module.fail_json(msg='reached timeout while waiting for rs.reconfig(): %s' % str(e))
            time.sleep(5)