根据索引冲突的报错信息 获取更新条件 Args: duplicate_errmsg: E11000 duplicate key error collection: feapder.test index: a_1_b_1 dup key: { : 1, : "你好" } data: {"a": 1, "b": "你好", "c": "嘻嘻"} Returns: {"a": 1, "b": "你好"}
(
self, coll_name: str, data: dict, duplicate_errmsg: str
)
| 468 | return index_keys |
| 469 | |
| 470 | def __get_update_condition( |
| 471 | self, coll_name: str, data: dict, duplicate_errmsg: str |
| 472 | ) -> dict: |
| 473 | """ |
| 474 | 根据索引冲突的报错信息 获取更新条件 |
| 475 | Args: |
| 476 | duplicate_errmsg: E11000 duplicate key error collection: feapder.test index: a_1_b_1 dup key: { : 1, : "你好" } |
| 477 | data: {"a": 1, "b": "你好", "c": "嘻嘻"} |
| 478 | |
| 479 | Returns: {"a": 1, "b": "你好"} |
| 480 | |
| 481 | """ |
| 482 | index_name = re.search(r"index: (\w+)", duplicate_errmsg).group(1) |
| 483 | index_keys = self.get_index_key(coll_name, index_name) |
| 484 | |
| 485 | condition = {key: data.get(key) for key in index_keys} |
| 486 | return condition |
| 487 | |
| 488 | def __getattr__(self, name): |
| 489 | return getattr(self.db, name) |
no test coverage detected