Spaces:
Running
Running
File size: 3,017 Bytes
acd7cf4 fb9c306 acd7cf4 fb9c306 acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 fb9c306 9e67c3b fb9c306 9e67c3b fb9c306 9e67c3b fb9c306 9e67c3b fb9c306 9e67c3b fb9c306 acd7cf4 fb9c306 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b 283e483 9e67c3b acd7cf4 9e67c3b acd7cf4 9e67c3b acd7cf4 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 |
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 98 99 100 101 102 103 104 105 106 107 108 |
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 BaseListStorage(Generic[T], StorageNameSpace):
def all_items(self) -> list[T]:
raise NotImplementedError
def get_by_index(self, index: int) -> Union[T, None]:
raise NotImplementedError
def append(self, data: T):
raise NotImplementedError
def upsert(self, data: list[T]):
raise NotImplementedError
def drop(self):
raise NotImplementedError
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
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
|