Merge a write command result into the full bulk result.
(
run: _Run,
full_result: MutableMapping[str, Any],
offset: int,
result: Mapping[str, Any],
)
| 70 | |
| 71 | |
| 72 | def _merge_command( |
| 73 | run: _Run, |
| 74 | full_result: MutableMapping[str, Any], |
| 75 | offset: int, |
| 76 | result: Mapping[str, Any], |
| 77 | ) -> None: |
| 78 | """Merge a write command result into the full bulk result.""" |
| 79 | affected = result.get("n", 0) |
| 80 | |
| 81 | if run.op_type == _INSERT: |
| 82 | full_result["nInserted"] += affected |
| 83 | |
| 84 | elif run.op_type == _DELETE: |
| 85 | full_result["nRemoved"] += affected |
| 86 | |
| 87 | elif run.op_type == _UPDATE: |
| 88 | upserted = result.get("upserted") |
| 89 | if upserted: |
| 90 | n_upserted = len(upserted) |
| 91 | for doc in upserted: |
| 92 | doc["index"] = run.index(doc["index"] + offset) |
| 93 | full_result["upserted"].extend(upserted) |
| 94 | full_result["nUpserted"] += n_upserted |
| 95 | full_result["nMatched"] += affected - n_upserted |
| 96 | else: |
| 97 | full_result["nMatched"] += affected |
| 98 | full_result["nModified"] += result["nModified"] |
| 99 | |
| 100 | write_errors = result.get("writeErrors") |
| 101 | if write_errors: |
| 102 | for doc in write_errors: |
| 103 | # Leave the server response intact for APM. |
| 104 | replacement = doc.copy() |
| 105 | idx = doc["index"] + offset |
| 106 | replacement["index"] = run.index(idx) |
| 107 | # Add the failed operation to the error document. |
| 108 | replacement["op"] = run.ops[idx] |
| 109 | full_result["writeErrors"].append(replacement) |
| 110 | |
| 111 | wce = _get_wce_doc(result) |
| 112 | if wce: |
| 113 | full_result["writeConcernErrors"].append(wce) |
| 114 | |
| 115 | |
| 116 | def _raise_bulk_write_error(full_result: _DocumentOut) -> NoReturn: |
no test coverage detected