Method _create_indexes

pymongo/synchronous/collection.py:2226–2273

Internal createIndexes helper. :param indexes: A list of :class:`~pymongo.operations.IndexModel` instances. :param session: a :class:`~pymongo.client_session.ClientSession`. :param kwargs: optional arguments to the createIndexes command (l

(
        self, indexes: Sequence[IndexModel], session: Optional[ClientSession], **kwargs: Any
    )

    @_csot.apply
    def _create_indexes(
        self, indexes: Sequence[IndexModel], session: Optional[ClientSession], **kwargs: Any
    ) -> list[str]:
        """Internal createIndexes helper.

        :param indexes: A list of :class:`~pymongo.operations.IndexModel`
            instances.
        :param session: a
            :class:`~pymongo.client_session.ClientSession`.
        :param kwargs: optional arguments to the createIndexes
            command (like maxTimeMS) can be passed as keyword arguments.
        """
        names = []

        def inner(
            session: Optional[ClientSession], conn: Connection, _retryable_write: bool
        ) -> list[str]:
            supports_quorum = conn.max_wire_version >= 9

            def gen_indexes() -> Iterator[Mapping[str, Any]]:
                for index in indexes:
                    if not isinstance(index, IndexModel):
                        raise TypeError(
                            f"{index!r} is not an instance of pymongo.operations.IndexModel"
                        )
                    document = index.document
                    names.append(document["name"])
                    yield document

            cmd = {"createIndexes": self.name, "indexes": list(gen_indexes())}
            cmd.update(kwargs)
            if "commitQuorum" in kwargs and not supports_quorum:
                raise ConfigurationError(
                    "Must be connected to MongoDB 4.4+ to use the "
                    "commitQuorum option for createIndexes"
                )

            self._command(
                conn,
                cmd,
                read_preference=ReadPreference.PRIMARY,
                codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
                write_concern=self._write_concern_for(session),
                session=session,
            )
            return names

        return self.database.client._retryable_write(False, inner, session, _Op.CREATE_INDEXES)

    def create_index(
        self,