# Error handling
if upsert_multi_result.results:
print(f"True if all bulk upserts succeeded, false otherwise: {upsert_multi_result.all_ok}")
else:
for couchbase_multi_upser_keys, couchbase_multi_upsert_exceptions in upsert_multi_result.exceptions.items():
print(f"{couchbase_multi_upser_keys} : {couchbase_multi_upsert_exceptions}")
Hi @ii – I would like to caution you on using MultiMutaitonResult’s results property as your check for if all the upsert operations succeeded from your multi upsert operation. Using if upsert_multi_result.results will be true if the results property is a non-empty dict (so it will be true if only a single operation succeeded an the rest failed). A better check would be to use the all_okay property or verify that every result in the results dict is a MutationResult: all(map(lambda r: isinstance(r, MutationResult), res.results.values()))
def upsert_records(cb_coll: Collection, records: Dict[str, dict]) -> Optional[Dict[str, Exception]]:
"""
Upsert multiple records into a Couchbase collection.
Args:
cb_coll (couchbase.collection.Collection): The Couchbase collection where records will be upserted.
records (Dict[str, dict]): A dictionary of records to upsert, where the keys are record identifiers.
Returns:
Optional[Dict[str, Exception]]: A dictionary of exceptions for records that failed to upsert, or None if all records were upserted successfully.
"""
upsert_multi_result = cb_coll.upsert_multi(records)
if not upsert_multi_result.all_ok:
return upsert_multi_result.exceptions
return None