(write_errors: list[Any])
| 285 | |
| 286 | |
| 287 | def _raise_last_write_error(write_errors: list[Any]) -> NoReturn: |
| 288 | # If the last batch had multiple errors only report |
| 289 | # the last error to emulate continue_on_error. |
| 290 | error = write_errors[-1] |
| 291 | if error.get("code") == 11000: |
| 292 | raise DuplicateKeyError(error.get("errmsg"), 11000, error) |
| 293 | raise WriteError(error.get("errmsg"), error.get("code"), error) |
| 294 | |
| 295 | |
| 296 | def _raise_write_concern_error(error: Any) -> NoReturn: |
no test coverage detected