Class InsertOne

pymongo/operations.py:85–135

Represents an insert_one operation.

class InsertOne(Generic[_DocumentType]):
    """Represents an insert_one operation."""

    __slots__ = (
        "_doc",
        "_namespace",
    )

    def __init__(self, document: _DocumentType, namespace: Optional[str] = None) -> None:
        """Create an InsertOne instance.

        For use with :meth:`~pymongo.asynchronous.collection.AsyncCollection.bulk_write`, :meth:`~pymongo.collection.Collection.bulk_write`,
        :meth:`~pymongo.asynchronous.mongo_client.AsyncMongoClient.bulk_write` and :meth:`~pymongo.mongo_client.MongoClient.bulk_write`.

        :param document: The document to insert. If the document is missing an
            _id field one will be added.
        :param namespace: (optional) The namespace in which to insert a document.

        .. versionchanged:: 4.9
           Added the `namespace` option to support `MongoClient.bulk_write`.
        """
        self._doc = document
        self._namespace = namespace

    def _add_to_bulk(self, bulkobj: _AgnosticBulk) -> None:
        """Add this operation to the _AsyncBulk/_Bulk instance `bulkobj`."""
        bulkobj.add_insert(self._doc)  # type: ignore[arg-type]

    def _add_to_client_bulk(self, bulkobj: _AgnosticClientBulk) -> None:
        """Add this operation to the _AsyncClientBulk/_ClientBulk instance `bulkobj`."""
        if not self._namespace:
            raise InvalidOperation(
                "MongoClient.bulk_write requires a namespace to be provided for each write operation"
            )
        bulkobj.add_insert(
            self._namespace,
            self._doc,  # type: ignore[arg-type]
        )

    def __repr__(self) -> str:
        if self._namespace:
            return f"{self.__class__.__name__}({self._doc!r}, {self._namespace!r})"
        return f"{self.__class__.__name__}({self._doc!r})"

    def __eq__(self, other: Any) -> bool:
        if type(other) == type(self):
            return other._doc == self._doc and other._namespace == self._namespace
        return NotImplemented

    def __ne__(self, other: Any) -> bool:
        return not self == other


class _DeleteOp: