An object wrapper for client-level bulk write API results.
| 287 | |
| 288 | |
| 289 | class ClientBulkWriteResult(_BulkWriteResultBase): |
| 290 | """An object wrapper for client-level bulk write API results.""" |
| 291 | |
| 292 | __slots__ = ("__has_verbose_results",) |
| 293 | |
| 294 | def __init__( |
| 295 | self, |
| 296 | bulk_api_result: MutableMapping[str, Any], |
| 297 | acknowledged: bool, |
| 298 | has_verbose_results: bool, |
| 299 | ) -> None: |
| 300 | """Create a ClientBulkWriteResult instance. |
| 301 | |
| 302 | :param bulk_api_result: A result dict from the client-level bulk write API |
| 303 | :param acknowledged: Was this write result acknowledged? If ``False`` |
| 304 | then all properties of this object will raise |
| 305 | :exc:`~pymongo.errors.InvalidOperation`. |
| 306 | :param has_verbose_results: Should the returned result be verbose? |
| 307 | If ``False``, then the ``insert_results``, ``update_results``, and |
| 308 | ``delete_results`` properties of this object will raise |
| 309 | :exc:`~pymongo.errors.InvalidOperation`. |
| 310 | """ |
| 311 | self.__has_verbose_results = has_verbose_results |
| 312 | super().__init__( |
| 313 | bulk_api_result, # type: ignore[arg-type] |
| 314 | acknowledged, |
| 315 | ) |
| 316 | |
| 317 | def __repr__(self) -> str: |
| 318 | return "{}({!r}, acknowledged={}, verbose={})".format( |
| 319 | self.__class__.__name__, |
| 320 | self.bulk_api_result, |
| 321 | self.acknowledged, |
| 322 | self.has_verbose_results, |
| 323 | ) |
| 324 | |
| 325 | def _raise_if_not_verbose(self, property_name: str) -> None: |
| 326 | """Raise an exception on property access if verbose results are off.""" |
| 327 | if not self.__has_verbose_results: |
| 328 | raise InvalidOperation( |
| 329 | f"A value for {property_name} is not available when " |
| 330 | "the results are not set to be verbose. Check the " |
| 331 | "verbose_results attribute to avoid this error." |
| 332 | ) |
| 333 | |
| 334 | @property |
| 335 | def has_verbose_results(self) -> bool: |
| 336 | """Whether the returned results should be verbose.""" |
| 337 | return self.__has_verbose_results |
| 338 | |
| 339 | @property |
| 340 | def insert_results(self) -> Mapping[int, InsertOneResult]: |
| 341 | """A map of successful insertion operations to their results.""" |
| 342 | self._raise_if_unacknowledged("insert_results") |
| 343 | self._raise_if_not_verbose("insert_results") |
| 344 | return cast( |
| 345 | Mapping[int, InsertOneResult], |
| 346 | self.bulk_api_result.get("insertResults"), |
no outgoing calls
no test coverage detected