The private guts of the bulk write API.
| 81 | |
| 82 | |
| 83 | class _Bulk: |
| 84 | """The private guts of the bulk write API.""" |
| 85 | |
| 86 | def __init__( |
| 87 | self, |
| 88 | collection: Collection[_DocumentType], |
| 89 | ordered: bool, |
| 90 | bypass_document_validation: Optional[bool], |
| 91 | comment: Optional[str] = None, |
| 92 | let: Optional[Any] = None, |
| 93 | ) -> None: |
| 94 | """Initialize a _Bulk instance.""" |
| 95 | self.collection = collection.with_options( |
| 96 | codec_options=collection.codec_options._replace( |
| 97 | unicode_decode_error_handler="replace", document_class=dict |
| 98 | ) |
| 99 | ) |
| 100 | self.let = let |
| 101 | if self.let is not None: |
| 102 | common.validate_is_document_type("let", self.let) |
| 103 | self.comment: Optional[str] = comment |
| 104 | self.ordered = ordered |
| 105 | self.ops: list[tuple[int, Mapping[str, Any]]] = [] |
| 106 | self.executed = False |
| 107 | self.bypass_doc_val = bypass_document_validation |
| 108 | self.uses_collation = False |
| 109 | self.uses_array_filters = False |
| 110 | self.uses_hint_update = False |
| 111 | self.uses_hint_delete = False |
| 112 | self.uses_sort = False |
| 113 | self.is_retryable = True |
| 114 | self.retrying = False |
| 115 | self.started_retryable_write = False |
| 116 | # Extra state so that we know where to pick up on a retry attempt. |
| 117 | self.current_run = None |
| 118 | self.next_run = None |
| 119 | self.is_encrypted = False |
| 120 | |
| 121 | @property |
| 122 | def bulk_ctx_class(self) -> Type[_BulkWriteContext]: |
| 123 | encrypter = self.collection.database.client._encrypter |
| 124 | if encrypter and not encrypter._bypass_auto_encryption: |
| 125 | self.is_encrypted = True |
| 126 | return _EncryptedBulkWriteContext |
| 127 | else: |
| 128 | self.is_encrypted = False |
| 129 | return _BulkWriteContext |
| 130 | |
| 131 | def add_insert(self, document: _DocumentOut) -> None: |
| 132 | """Add an insert document to the list of ops.""" |
| 133 | validate_is_document_type("document", document) |
| 134 | # Generate ObjectId client side. |
| 135 | if not (isinstance(document, RawBSONDocument) or "_id" in document): |
| 136 | document["_id"] = ObjectId() |
| 137 | self.ops.append((_INSERT, document)) |
| 138 | |
| 139 | def add_update( |
| 140 | self, |
no outgoing calls
no test coverage detected