The private guts of the client-level bulk write API.
| 85 | |
| 86 | |
| 87 | class _ClientBulk: |
| 88 | """The private guts of the client-level bulk write API.""" |
| 89 | |
| 90 | def __init__( |
| 91 | self, |
| 92 | client: MongoClient[Any], |
| 93 | write_concern: WriteConcern, |
| 94 | ordered: bool = True, |
| 95 | bypass_document_validation: Optional[bool] = None, |
| 96 | comment: Optional[str] = None, |
| 97 | let: Optional[Any] = None, |
| 98 | verbose_results: bool = False, |
| 99 | ) -> None: |
| 100 | """Initialize a _ClientBulk instance.""" |
| 101 | self.client = client |
| 102 | self.write_concern = write_concern |
| 103 | self.let = let |
| 104 | if self.let is not None: |
| 105 | common.validate_is_document_type("let", self.let) |
| 106 | self.ordered = ordered |
| 107 | self.bypass_doc_val = bypass_document_validation |
| 108 | self.comment = comment |
| 109 | self.verbose_results = verbose_results |
| 110 | self.ops: list[tuple[str, Mapping[str, Any]]] = [] |
| 111 | self.namespaces: list[str] = [] |
| 112 | self.idx_offset: int = 0 |
| 113 | self.total_ops: int = 0 |
| 114 | self.executed = False |
| 115 | self.uses_collation = False |
| 116 | self.uses_array_filters = False |
| 117 | self.is_retryable = self.client.options.retry_writes |
| 118 | self.retrying = False |
| 119 | self.started_retryable_write = False |
| 120 | |
| 121 | @property |
| 122 | def bulk_ctx_class(self) -> Type[_ClientBulkWriteContext]: |
| 123 | return _ClientBulkWriteContext |
| 124 | |
| 125 | def add_insert(self, namespace: str, document: _DocumentOut) -> None: |
| 126 | """Add an insert document to the list of ops.""" |
| 127 | validate_is_document_type("document", document) |
| 128 | # Generate ObjectId client side. |
| 129 | if not (isinstance(document, RawBSONDocument) or "_id" in document): |
| 130 | document["_id"] = ObjectId() |
| 131 | cmd = {"insert": -1, "document": document} |
| 132 | self.ops.append(("insert", cmd)) |
| 133 | self.namespaces.append(namespace) |
| 134 | self.total_ops += 1 |
| 135 | |
| 136 | def add_update( |
| 137 | self, |
| 138 | namespace: str, |
| 139 | selector: Mapping[str, Any], |
| 140 | update: Union[Mapping[str, Any], _Pipeline], |
| 141 | multi: bool, |
| 142 | upsert: Optional[bool] = None, |
| 143 | collation: Optional[Mapping[str, Any]] = None, |
| 144 | array_filters: Optional[list[Mapping[str, Any]]] = None, |