Coverage for src / lexigram / contracts / data / graph / types.py: 3%

1"""Graph store value types — all immutable.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from typing import TYPE_CHECKING, Any 

7 

8from lexigram.contracts.data.graph.enums import ( 

9 ConstraintKind, 

10 EdgeDirection, 

11 IndexKind, 

12 ReturnType, 

13) 

14 

15if TYPE_CHECKING: 

16 from lexigram.contracts.data.graph.filters import PropertyFilter 

17 

18 

19@dataclass(frozen=True, slots=True) 

20class GraphNode: 

21 """A node (vertex) in the graph.""" 

22 

23 id: str 

24 labels: tuple[str, ...] 

25 properties: dict[str, Any] = field(default_factory=dict) 

26 

27 

28@dataclass(frozen=True, slots=True) 

29class GraphEdge: 

30 """A directed edge (relationship) in the graph.""" 

31 

32 id: str 

33 type: str 

34 source_id: str 

35 target_id: str 

36 properties: dict[str, Any] = field(default_factory=dict) 

37 

38 

39@dataclass(frozen=True, slots=True) 

40class GraphPath: 

41 """An ordered alternating sequence of nodes and edges. 

42 

43 ``nodes[0] --edges[0]--> nodes[1] --edges[1]--> nodes[2] ...`` 

44 """ 

45 

46 nodes: tuple[GraphNode, ...] 

47 edges: tuple[GraphEdge, ...] 

48 

49 @property 

50 def length(self) -> int: 

51 """Number of edges (hops) in the path.""" 

52 return len(self.edges) 

53 

54 @property 

55 def start_node(self) -> GraphNode: 

56 """First node in the path.""" 

57 return self.nodes[0] 

58 

59 @property 

60 def end_node(self) -> GraphNode: 

61 """Last node in the path.""" 

62 return self.nodes[-1] 

63 

64 

65@dataclass(frozen=True, slots=True) 

66class NodeSpec: 

67 """Specification for creating a node.""" 

68 

69 labels: tuple[str, ...] 

70 properties: dict[str, Any] = field(default_factory=dict) 

71 id: str | None = None 

72 

73 

74@dataclass(frozen=True, slots=True) 

75class EdgeSpec: 

76 """Specification for creating an edge.""" 

77 

78 source_id: str 

79 target_id: str 

80 type: str 

81 properties: dict[str, Any] = field(default_factory=dict) 

82 

83 

84@dataclass(frozen=True, slots=True) 

85class NodeResult: 

86 """Result of a node creation or update.""" 

87 

88 id: str 

89 created: bool = True 

90 

91 

92@dataclass(frozen=True, slots=True) 

93class EdgeResult: 

94 """Result of an edge creation or update.""" 

95 

96 id: str 

97 created: bool = True 

98 

99 

100@dataclass(frozen=True, slots=True) 

101class BulkNodeResult: 

102 """Result of a bulk node operation.""" 

103 

104 created_count: int 

105 ids: tuple[str, ...] = () 

106 

107 

108@dataclass(frozen=True, slots=True) 

109class BulkEdgeResult: 

110 """Result of a bulk edge operation.""" 

111 

112 created_count: int 

113 ids: tuple[str, ...] = () 

114 

115 

116@dataclass(frozen=True, slots=True) 

117class GraphInfo: 

118 """Metadata about a graph database.""" 

119 

120 name: str 

121 node_count: int = 0 

122 edge_count: int = 0 

123 label_count: int = 0 

124 edge_type_count: int = 0 

125 

126 

127@dataclass(frozen=True, slots=True) 

128class StartSpec: 

129 """How to find traversal start nodes.""" 

130 

131 node_ids: tuple[str, ...] | None = None 

132 labels: tuple[str, ...] | None = None 

133 properties: dict[str, Any] | None = None 

134 filter: PropertyFilter | None = None 

135 

136 

137@dataclass(frozen=True, slots=True) 

138class TraversalStep: 

139 """A single hop definition in a traversal.""" 

140 

141 edge_types: tuple[str, ...] | None = None 

142 direction: EdgeDirection = EdgeDirection.OUTGOING 

143 min_depth: int = 1 

144 max_depth: int = 1 

145 node_labels: tuple[str, ...] | None = None 

146 node_filter: PropertyFilter | None = None 

147 edge_filter: PropertyFilter | None = None 

148 

149 

150@dataclass(frozen=True, slots=True) 

151class TraversalQuery: 

152 """Compiled traversal specification.""" 

153 

154 start: StartSpec 

155 steps: tuple[TraversalStep, ...] 

156 return_type: ReturnType = ReturnType.PATHS 

157 result_filter: PropertyFilter | None = None 

158 result_labels: tuple[str, ...] | None = None 

159 order_by: tuple[tuple[str, bool], ...] | None = None 

160 limit: int | None = None 

161 skip: int | None = None 

162 unique_nodes: bool = True 

163 

164 

165@dataclass(frozen=True, slots=True) 

166class IndexSpec: 

167 """Specification for creating a graph index.""" 

168 

169 name: str 

170 label: str 

171 properties: tuple[str, ...] 

172 kind: IndexKind = IndexKind.BTREE 

173 

174 

175@dataclass(frozen=True, slots=True) 

176class ConstraintSpec: 

177 """Specification for creating a graph constraint.""" 

178 

179 name: str 

180 label: str 

181 properties: tuple[str, ...] 

182 kind: ConstraintKind = ConstraintKind.UNIQUE 

183 

184 

185__all__ = [ 

186 "BulkEdgeResult", 

187 "BulkNodeResult", 

188 "ConstraintSpec", 

189 "EdgeResult", 

190 "EdgeSpec", 

191 "GraphEdge", 

192 "GraphInfo", 

193 "GraphNode", 

194 "GraphPath", 

195 "IndexSpec", 

196 "NodeResult", 

197 "NodeSpec", 

198 "StartSpec", 

199 "TraversalQuery", 

200 "TraversalStep", 

201]