获取参与索引的key Args: index_name: 索引名 Returns:
(self, coll_name, index_name)
| 444 | return self.db.drop_collection(coll_name) |
| 445 | |
| 446 | def get_index_key(self, coll_name, index_name): |
| 447 | """ |
| 448 | 获取参与索引的key |
| 449 | Args: |
| 450 | index_name: 索引名 |
| 451 | |
| 452 | Returns: |
| 453 | |
| 454 | """ |
| 455 | cache_key = f"{coll_name}:{index_name}" |
| 456 | |
| 457 | if cache_key in self.__index__cached: |
| 458 | return self.__index__cached.get(cache_key) |
| 459 | |
| 460 | index = self.get_index(coll_name) |
| 461 | index_detail = index.get(index_name) |
| 462 | if not index_detail: |
| 463 | errmsg = f"not found index {index_name} in collection {coll_name}" |
| 464 | raise Exception(errmsg) |
| 465 | |
| 466 | index_keys = [val[0] for val in index_detail.get("key")] |
| 467 | self.__index__cached[cache_key] = index_keys |
| 468 | return index_keys |
| 469 | |
| 470 | def __get_update_condition( |
| 471 | self, coll_name: str, data: dict, duplicate_errmsg: str |
no test coverage detected