Coverage for target/native-coverage/profiles/8cb89e7529e5488ca16399c0daa3ee58/project/.venv/lib/python3.13/site-packages/rdflib_turbo/_reads.py: 100%

1"""Typed, projected reads over graph-scoped native storage cursors.""" 

2 

3from rdflib import BNode, Literal, URIRef 

4from rdflib.paths import Path 

5 

6from ._literal import restore 

7from ._native import NativeInputTerms, NativeTermFactories 

8from .store import Store 

9 

10_TERM_FACTORIES = NativeTermFactories(URIRef, BNode, restore, str.__new__, True) 

11_INPUT_TERMS = NativeInputTerms(URIRef, BNode, Literal) 

12 

13 

14def encode(value): 

15 if value is None: 

16 return None 

17 if isinstance(value, URIRef): 

18 return ("uri", value, None, None) 

19 if isinstance(value, BNode): 

20 return ("bnode", value, None, None) 

21 if isinstance(value, Literal): 

22 return ("literal", value, value.datatype, value.language) 

23 raise TypeError(f"Unsupported RDF term: {type(value).__name__}") 

24 

25 

26def encode_graph(store, context): 

27 identifier = context.identifier if hasattr(context, "identifier") else context 

28 return None if identifier == store.default_identifier else encode(identifier) 

29 

30 

31def scoped_rows(store, pattern, context, positions=(0, 1, 2)): 

32 graph = encode_graph(store, context) 

33 # Bound terms already belong to the caller: transport/decode only unknown outputs. 

34 count = len(positions) 

35 first = positions[0] 

36 mask = 7 if count == 3 else (1 << first) | (1 << positions[-1]) 

37 output_mask = mask & ( 

38 (pattern[0] is None) | ((pattern[1] is None) << 1) | ((pattern[2] is None) << 2) 

39 ) 

40 batch, cursor = store.backend._native.read_pattern( 

41 (encode(pattern[0]), encode(pattern[1]), encode(pattern[2])), 

42 graph, 

43 output_mask, 

44 _TERM_FACTORIES, 

45 ) 

46 try: 

47 while True: 

48 if count == 1: 

49 for row in batch: 

50 yield row[first] 

51 elif count == 2: 

52 second = positions[1] 

53 for row in batch: 

54 yield row[first], row[second] 

55 elif output_mask == 7: 

56 # Every SPO slot is unbound: native tuples already contain the complete result. 

57 yield from batch 

58 else: 

59 for subject, predicate, obj in batch: 

60 yield ( 

61 pattern[0] if subject is None else subject, 

62 pattern[1] if predicate is None else predicate, 

63 pattern[2] if obj is None else obj, 

64 ) 

65 if cursor is None: 

66 break 

67 batch, finished = cursor.next_batch() 

68 if finished: 

69 cursor = None 

70 finally: 

71 if cursor is not None: 

72 cursor.close() 

73 

74 

75class ProjectedReads: 

76 """Avoid constructing unused RDF terms for ordinary graph projections.""" 

77 

78 def __contains__(self, pattern): 

79 if ( 

80 type(self).triples is not ProjectedReads.triples 

81 or type(self).__getattribute__ is not object.__getattribute__ 

82 or "triples" in self.__dict__ 

83 or not isinstance(self.store.backend, Store) 

84 or isinstance(pattern[1], Path) 

85 or pattern[0] is None 

86 or pattern[1] is None 

87 or pattern[2] is None 

88 ): 

89 return super().__contains__(pattern) 

90 return self.store.backend._native.contains_pattern(pattern, self, _INPUT_TERMS) 

91 

92 def triples(self, pattern): 

93 if isinstance(pattern[1], Path) or not isinstance(self.store.backend, Store): 

94 yield from super().triples(pattern) 

95 else: 

96 yield from scoped_rows(self.store, pattern, self) 

97 

98 def _project(self, pattern, positions): 

99 if isinstance(self.store.backend, Store): 

100 yield from scoped_rows(self.store, pattern, self, positions) 

101 return 

102 # Explicit transactions retain their owner-thread read protocol. 

103 rows = self.triples(pattern) 

104 try: 

105 for row in rows: 

106 yield row[positions[0]] if len(positions) == 1 else tuple(row[i] for i in positions) 

107 finally: 

108 rows.close() 

109 

110 def _custom_triples(self): 

111 return type(self).triples is not ProjectedReads.triples 

112 

113 def subjects(self, predicate=None, object=None, unique=False): 

114 if ( 

115 unique 

116 or isinstance(object, list) 

117 or isinstance(predicate, Path) 

118 or self._custom_triples() 

119 ): 

120 yield from super().subjects(predicate, object, unique) 

121 else: 

122 yield from self._project((None, predicate, object), (0,)) 

123 

124 def predicates(self, subject=None, object=None, unique=False): 

125 if unique or self._custom_triples(): 

126 yield from super().predicates(subject, object, unique) 

127 else: 

128 yield from self._project((subject, None, object), (1,)) 

129 

130 def objects(self, subject=None, predicate=None, unique=False): 

131 if ( 

132 unique 

133 or isinstance(subject, list) 

134 or isinstance(predicate, Path) 

135 or self._custom_triples() 

136 ): 

137 yield from super().objects(subject, predicate, unique) 

138 else: 

139 yield from self._project((subject, predicate, None), (2,)) 

140 

141 def subject_objects(self, predicate=None, unique=False): 

142 if unique or isinstance(predicate, Path) or self._custom_triples(): 

143 yield from super().subject_objects(predicate, unique) 

144 else: 

145 yield from self._project((None, predicate, None), (0, 2)) 

146 

147 def predicate_objects(self, subject=None, unique=False): 

148 if unique or self._custom_triples(): 

149 yield from super().predicate_objects(subject, unique) 

150 else: 

151 yield from self._project((subject, None, None), (1, 2)) 

152 

153 def subject_predicates(self, object=None, unique=False): 

154 if unique or self._custom_triples(): 

155 yield from super().subject_predicates(object, unique) 

156 else: 

157 yield from self._project((None, None, object), (0, 1))