Path:
strictdoc/core/graph_database.py
Lines:
120
Non-empty lines:
102
Non-empty lines covered with requirements:
102 / 102 (100.0%)
Functions:
16
Functions covered by requirements:
16 / 16 (100.0%)
1
"""2
@relation(SDOC-SRS-28, scope=file)3
"""4
5
from typing import Any, Dict, Hashable, List, Optional, Tuple
6
7
from strictdoc.core.graph.abstract_bucket import AbstractBucket
8
from strictdoc.helpers.ordered_set import OrderedSet
9
10
11
class ConstraintViolation(Exception):
12
def __init__(self, message: str) -> None:
13
super().__init__(message)
14
15
16
class GraphDatabase:
17
def __init__(self, buckets: List[Tuple[Any, AbstractBucket]]):
18
self._links = None
19
self._id_to_bucket: Dict[Any, AbstractBucket] = {}
20
for bucket_ in buckets:
21
assert bucket_[0] not in self._id_to_bucket
22
self._id_to_bucket[bucket_[0]] = bucket_[1]
23
24
def has_any_link(self, *, link_type: Hashable, lhs_node: Any) -> bool:
25
return self._id_to_bucket[link_type].has_any_link(lhs_node=lhs_node)
26
27
def has_link(
28
self,
29
*,
30
link_type: Hashable,
31
lhs_node: Any,
32
rhs_node: Any,
33
edge: Optional[str] = None,
34
) -> bool:
35
return self._id_to_bucket[link_type].has_link(
36
lhs_node=lhs_node, rhs_node=rhs_node, edge=edge
37
)38
39
def get_count(self, *, link_type: Hashable) -> int:
40
return self._id_to_bucket[link_type].get_count()
41
42
def get_link_value_weak(
43
self, *, link_type: Hashable, lhs_node: Any
44
) -> Optional[Any]:
45
return self._id_to_bucket[link_type].get_link_value_weak(
46
lhs_node=lhs_node
47
)48
49
def get_link_value(
50
self, *, link_type: Hashable, lhs_node: Any, edge: Optional[str] = None
51
) -> Any:
52
return self._id_to_bucket[link_type].get_link_value(
53
lhs_node=lhs_node, edge=edge
54
)55
56
def get_link_values(
57
self, *, link_type: Hashable, lhs_node: Any, edge: Optional[str] = None
58
) -> OrderedSet[Any]:
59
return self._id_to_bucket[link_type].get_link_values(
60
lhs_node=lhs_node, edge=edge
61
)62
63
def get_link_values_with_edges(
64
self, *, link_type: Hashable, lhs_node: Any, edge: Optional[str] = None
65
) -> List[Tuple[Any, Optional[str]]]:
66
return self._id_to_bucket[link_type].get_link_values_with_edges(
67
lhs_node=lhs_node, edge=edge
68
)69
70
def get_link_values_reverse(
71
self, *, link_type: Hashable, rhs_node: Any
72
) -> Any:
73
return self._id_to_bucket[link_type].get_link_values_reverse(
74
rhs_node=rhs_node
75
)76
77
def create_link(
78
self,
79
*,
80
link_type: Hashable,
81
lhs_node: Any,
82
rhs_node: Any,
83
edge: Optional[str] = None,
84
) -> None:
85
assert lhs_node is not None, lhs_node
86
assert rhs_node is not None, rhs_node
87
self._id_to_bucket[link_type].create_link(
88
lhs_node=lhs_node, rhs_node=rhs_node, edge=edge
89
)90
91
def delete_link(
92
self,
93
*,
94
link_type: Hashable,
95
lhs_node: Any,
96
rhs_node: Any,
97
edge: Optional[str] = None,
98
) -> None:
99
self._id_to_bucket[link_type].delete_link(
100
lhs_node=lhs_node, rhs_node=rhs_node, edge=edge
101
)102
103
def delete_link_weak(
104
self,
105
*,
106
link_type: Hashable,
107
lhs_node: Any,
108
rhs_node: Any,
109
) -> None:
110
self._id_to_bucket[link_type].delete_link_weak(
111
lhs_node=lhs_node, rhs_node=rhs_node
112
)113
114
def delete_all_links(
115
self,
116
*,
117
link_type: Hashable,
118
lhs_node: Any,
119
) -> None:
120
self._id_to_bucket[link_type].delete_all_links(lhs_node=lhs_node)