(
self,
target: Union[
MongoClient[_DocumentType],
Database[_DocumentType],
Collection[_DocumentType],
],
pipeline: Optional[_Pipeline],
full_document: Optional[str],
resume_after: Optional[Mapping[str, Any]],
max_await_time_ms: Optional[int],
batch_size: Optional[int],
collation: Optional[_CollationIn],
start_at_operation_time: Optional[Timestamp],
session: Optional[ClientSession],
start_after: Optional[Mapping[str, Any]],
comment: Optional[Any] = None,
full_document_before_change: Optional[str] = None,
show_expanded_events: Optional[bool] = None,
)
| 100 | """ |
| 101 | |
| 102 | def __init__( |
| 103 | self, |
| 104 | target: Union[ |
| 105 | MongoClient[_DocumentType], |
| 106 | Database[_DocumentType], |
| 107 | Collection[_DocumentType], |
| 108 | ], |
| 109 | pipeline: Optional[_Pipeline], |
| 110 | full_document: Optional[str], |
| 111 | resume_after: Optional[Mapping[str, Any]], |
| 112 | max_await_time_ms: Optional[int], |
| 113 | batch_size: Optional[int], |
| 114 | collation: Optional[_CollationIn], |
| 115 | start_at_operation_time: Optional[Timestamp], |
| 116 | session: Optional[ClientSession], |
| 117 | start_after: Optional[Mapping[str, Any]], |
| 118 | comment: Optional[Any] = None, |
| 119 | full_document_before_change: Optional[str] = None, |
| 120 | show_expanded_events: Optional[bool] = None, |
| 121 | ) -> None: |
| 122 | if pipeline is None: |
| 123 | pipeline = [] |
| 124 | pipeline = common.validate_list("pipeline", pipeline) |
| 125 | common.validate_string_or_none("full_document", full_document) |
| 126 | validate_collation_or_none(collation) |
| 127 | common.validate_non_negative_integer_or_none("batchSize", batch_size) |
| 128 | |
| 129 | self._decode_custom = False |
| 130 | self._orig_codec_options: CodecOptions[_DocumentType] = target.codec_options |
| 131 | if target.codec_options.type_registry._decoder_map: |
| 132 | self._decode_custom = True |
| 133 | # Keep the type registry so that we support encoding custom types |
| 134 | # in the pipeline. |
| 135 | self._target = target.with_options( # type: ignore |
| 136 | codec_options=target.codec_options.with_options(document_class=RawBSONDocument) |
| 137 | ) |
| 138 | else: |
| 139 | self._target = target |
| 140 | |
| 141 | self._pipeline = copy.deepcopy(pipeline) |
| 142 | self._full_document = full_document |
| 143 | self._full_document_before_change = full_document_before_change |
| 144 | self._uses_start_after = start_after is not None |
| 145 | self._uses_resume_after = resume_after is not None |
| 146 | self._resume_token = copy.deepcopy(start_after or resume_after) |
| 147 | self._max_await_time_ms = max_await_time_ms |
| 148 | self._batch_size = batch_size |
| 149 | self._collation = collation |
| 150 | self._start_at_operation_time = start_at_operation_time |
| 151 | self._session = session |
| 152 | self._comment = comment |
| 153 | self._closed = False |
| 154 | self._timeout = self._target._timeout |
| 155 | self._show_expanded_events = show_expanded_events |
| 156 | |
| 157 | def _initialize_cursor(self) -> None: |
| 158 | # Initialize cursor. |
nothing calls this directly
no test coverage detected