Create an update document and add it to the list of ops.
(
self,
namespace: str,
selector: Mapping[str, Any],
update: Union[Mapping[str, Any], _Pipeline],
multi: bool,
upsert: Optional[bool] = None,
collation: Optional[Mapping[str, Any]] = None,
array_filters: Optional[list[Mapping[str, Any]]] = None,
hint: Union[str, dict[str, Any], None] = None,
sort: Optional[Mapping[str, Any]] = None,
)
| 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, |
| 145 | hint: Union[str, dict[str, Any], None] = None, |
| 146 | sort: Optional[Mapping[str, Any]] = None, |
| 147 | ) -> None: |
| 148 | """Create an update document and add it to the list of ops.""" |
| 149 | validate_ok_for_update(update) |
| 150 | cmd = { |
| 151 | "update": -1, |
| 152 | "filter": selector, |
| 153 | "updateMods": update, |
| 154 | "multi": multi, |
| 155 | } |
| 156 | if upsert is not None: |
| 157 | cmd["upsert"] = upsert |
| 158 | if array_filters is not None: |
| 159 | self.uses_array_filters = True |
| 160 | cmd["arrayFilters"] = array_filters |
| 161 | if hint is not None: |
| 162 | cmd["hint"] = hint |
| 163 | if collation is not None: |
| 164 | self.uses_collation = True |
| 165 | cmd["collation"] = collation |
| 166 | if sort is not None: |
| 167 | cmd["sort"] = sort |
| 168 | if multi: |
| 169 | # A bulk_write containing an update_many is not retryable. |
| 170 | self.is_retryable = False |
| 171 | self.ops.append(("update", cmd)) |
| 172 | self.namespaces.append(namespace) |
| 173 | self.total_ops += 1 |
| 174 | |
| 175 | def add_replace( |
| 176 | self, |
nothing calls this directly
no test coverage detected