Spaces:
Running
Running
File size: 2,720 Bytes
acd7cf4 fb9c306 acd7cf4 fb9c306 acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 fb9c306 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b 283e483 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 31086ae fb9c306 acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b fb9c306 acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 31086ae |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
from dataclasses import dataclass
from typing import Generic, TypeVar, Union
T = TypeVar("T")
@dataclass
class StorageNameSpace:
working_dir: str = None
namespace: str = None
def index_done_callback(self):
"""commit the storage operations after indexing"""
def query_done_callback(self):
"""commit the storage operations after querying"""
class BaseKVStorage(Generic[T], StorageNameSpace):
def all_keys(self) -> list[str]:
raise NotImplementedError
def get_by_id(self, id: str) -> Union[T, None]:
raise NotImplementedError
def get_by_ids(
self, ids: list[str], fields: Union[set[str], None] = None
) -> list[Union[T, None]]:
raise NotImplementedError
def get_all(self) -> dict[str, T]:
raise NotImplementedError
def filter_keys(self, data: list[str]) -> set[str]:
"""return un-exist keys"""
raise NotImplementedError
def upsert(self, data: dict[str, T]):
raise NotImplementedError
def drop(self):
raise NotImplementedError
def reload(self):
raise NotImplementedError
class BaseGraphStorage(StorageNameSpace):
def has_node(self, node_id: str) -> bool:
raise NotImplementedError
def has_edge(self, source_node_id: str, target_node_id: str) -> bool:
raise NotImplementedError
def node_degree(self, node_id: str) -> int:
raise NotImplementedError
def edge_degree(self, src_id: str, tgt_id: str) -> int:
raise NotImplementedError
def get_node(self, node_id: str) -> Union[dict, None]:
raise NotImplementedError
def update_node(self, node_id: str, node_data: dict[str, str]):
raise NotImplementedError
def get_all_nodes(self) -> Union[list[tuple[str, dict]], None]:
raise NotImplementedError
def get_edge(self, source_node_id: str, target_node_id: str) -> Union[dict, None]:
raise NotImplementedError
def update_edge(
self, source_node_id: str, target_node_id: str, edge_data: dict[str, str]
):
raise NotImplementedError
def get_all_edges(self) -> Union[list[tuple[str, str, dict]], None]:
raise NotImplementedError
def get_node_edges(self, source_node_id: str) -> Union[list[tuple[str, str]], None]:
raise NotImplementedError
def upsert_node(self, node_id: str, node_data: dict[str, str]):
raise NotImplementedError
def upsert_edge(
self, source_node_id: str, target_node_id: str, edge_data: dict[str, str]
):
raise NotImplementedError
def delete_node(self, node_id: str):
raise NotImplementedError
def reload(self):
raise NotImplementedError
|