Initialize a _Bulk instance.
(
self,
collection: Collection[_DocumentType],
ordered: bool,
bypass_document_validation: Optional[bool],
comment: Optional[str] = None,
let: Optional[Any] = None,
)
| 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]: |
nothing calls this directly
no test coverage detected