Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/record.py: 13%

1import json 

2import typing as t 

3 

4from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity 

5 

6if t.TYPE_CHECKING: 6 ↛ 7line 6 didn't jump to line 7 because the condition on line 6 was never true

7 from ..skeleton import SkeletonInstance 

8 

9 

10class RecordBone(BaseBone): 

11 """ 

12 The RecordBone class is a specialized bone type used to store structured data. It inherits from 

13 the BaseBone class. The RecordBone class is designed to store complex data structures, such as 

14 nested dictionaries or objects, by using a related skeleton class (the using parameter) to manage 

15 the internal structure of the data. 

16 

17 :param format: Optional string parameter to specify the format of the record bone. 

18 :param indexed: Optional boolean parameter to indicate if the record bone is indexed. 

19 Defaults to False. 

20 :param using: A class that inherits from 'viur.core.skeleton.RelSkel' to be used with the 

21 RecordBone. 

22 :param kwargs: Additional keyword arguments to be passed to the BaseBone constructor. 

23 """ 

24 type = "record" 

25 

26 def __init__( 

27 self, 

28 *, 

29 format: str = None, 

30 indexed: bool = False, 

31 using: 'viur.core.skeleton.RelSkel' = None, 

32 **kwargs 

33 ): 

34 from viur.core.skeleton import RelSkel 

35 if not issubclass(using, RelSkel): 

36 raise ValueError("RecordBone requires for valid using-parameter (subclass of viur.core.skeleton.RelSkel)") 

37 

38 super().__init__(indexed=indexed, **kwargs) 

39 self.using = using 

40 self.format = format 

41 if not format or indexed: 

42 raise NotImplementedError("A RecordBone must not be indexed and must have a format set") 

43 

44 def singleValueUnserialize(self, val): 

45 """ 

46 Unserializes a single value, creating an instance of the 'using' class and unserializing 

47 the value into it. 

48 

49 :param val: The value to unserialize. 

50 :return: An instance of the 'using' class with the unserialized data. 

51 :raises AssertionError: If the unserialized value is not a dictionary. 

52 """ 

53 if isinstance(val, str): 

54 try: 

55 value = json.loads(val) 

56 except ValueError: 

57 value = None 

58 else: 

59 value = val 

60 

61 if not value: 

62 return None 

63 

64 if isinstance(value, list) and value: 

65 value = value[0] 

66 

67 assert isinstance(value, dict), f"Read something from the datastore thats not a dict: {type(value)}" 

68 

69 usingSkel = self.using() 

70 usingSkel.unserialize(value) 

71 return usingSkel 

72 

73 def singleValueSerialize(self, value, skel: 'SkeletonInstance', name: str, parentIndexed: bool): 

74 """ 

75 Serializes a single value by calling the serialize method of the 'using' skeleton instance. 

76 

77 :param value: The value to be serialized, which should be an instance of the 'using' skeleton. 

78 :param skel: The parent skeleton instance. 

79 :param name: The name of the bone. 

80 :param parentIndexed: A boolean indicating if the parent bone is indexed. 

81 :return: The serialized value. 

82 """ 

83 if not value: 

84 return value 

85 

86 return value.serialize(parentIndexed=False) 

87 

88 def _get_single_destinct_hash(self, value): 

89 return tuple(bone._get_destinct_hash(value[name]) for name, bone in self.using.__boneMap__.items()) 

90 

91 def parseSubfieldsFromClient(self) -> bool: 

92 """ 

93 Determines if the current request should attempt to parse subfields received from the client. 

94 This should only be set to True if a list of dictionaries is expected to be transmitted. 

95 """ 

96 return True 

97 

98 def singleValueFromClient(self, value, skel, bone_name, client_data): 

99 usingSkel = self.using() 

100 if not usingSkel.fromClient(value): 

101 usingSkel.errors.append( 

102 ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Incomplete data") 

103 ) 

104 return usingSkel, usingSkel.errors 

105 

106 def postSavedHandler(self, skel, boneName, key) -> None: 

107 super().postSavedHandler(skel, boneName, key) 

108 for idx, lang, value in self.iter_bone_value(skel, boneName): 

109 using = self.using() 

110 using.unserialize(value) 

111 for bone_name, bone in using.items(): 

112 bone.postSavedHandler(using, bone_name, None) 

113 

114 def refresh(self, skel, boneName) -> None: 

115 super().refresh(skel, boneName) 

116 for idx, lang, value in self.iter_bone_value(skel, boneName): 

117 using = self.using() 

118 using.unserialize(value) 

119 for bone_name, bone in using.items(): 

120 bone.refresh(using, bone_name) 

121 

122 def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: 

123 """ 

124 Collects search tags from the 'using' skeleton instance for the given bone. 

125 

126 :param skel: The parent skeleton instance. 

127 :param name: The name of the bone. 

128 :return: A set of search tags generated from the 'using' skeleton instance. 

129 """ 

130 result = set() 

131 

132 using_skel_cache = self.using() 

133 for idx, lang, value in self.iter_bone_value(skel, name): 

134 if value is None: 

135 continue 

136 for key, bone in using_skel_cache.items(): 

137 if not bone.searchable: 

138 continue 

139 for tag in bone.getSearchTags(value, key): 

140 result.add(tag) 

141 

142 return result 

143 

144 def getSearchDocumentFields(self, valuesCache, name, prefix=""): 

145 """ 

146 Generates a list of search document fields for the given values cache, name, and optional prefix. 

147 

148 :param dict valuesCache: A dictionary containing the cached values. 

149 :param str name: The name of the bone to process. 

150 :param str prefix: An optional prefix to use for the search document fields, defaults to an empty string. 

151 :return: A list of search document fields. 

152 :rtype: list 

153 """ 

154 

155 def getValues(res, skel, valuesCache, searchPrefix): 

156 for key, bone in skel.items(): 

157 if bone.searchable: 

158 res.extend(bone.getSearchDocumentFields(valuesCache, key, prefix=searchPrefix)) 

159 

160 value = valuesCache.get(name) 

161 res = [] 

162 

163 if not value: 

164 return res 

165 uskel = self.using() 

166 for idx, val in enumerate(value): 

167 getValues(res, uskel, val, f"{prefix}{name}_{idx}") 

168 

169 return res 

170 

171 def getReferencedBlobs(self, skel: "SkeletonInstance", name: str) -> set[str]: 

172 """ 

173 Retrieves a set of referenced blobs for the given skeleton instance and name. 

174 

175 :param skel: The skeleton instance to process. 

176 :param name: The name of the bone to process. 

177 :return: A set of referenced blobs. 

178 """ 

179 result = set() 

180 

181 using_skel_cache = self.using() 

182 for idx, lang, value in self.iter_bone_value(skel, name): 

183 if value is None: 

184 continue 

185 for key, bone in using_skel_cache.items(): 

186 result |= bone.getReferencedBlobs(value, key) 

187 

188 return result 

189 

190 def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> list[str]: 

191 """ 

192 This method is intentionally not implemented as it's not possible to determine how to derive 

193 a key from the related skeleton being used (i.e., which fields to include and how). 

194 

195 """ 

196 raise NotImplementedError() 

197 

198 def structure(self) -> dict: 

199 return super().structure() | { 

200 "format": self.format, 

201 "using": self.using().structure()} 

202 

203 def refresh(self, skel, bone_name): 

204 using_skel = self.using() 

205 

206 for idx, lang, value in self.iter_bone_value(skel, bone_name): 

207 if value is None: 

208 continue 

209 

210 using_skel.unserialize(value) 

211 

212 for key, bone in using_skel.items(): 

213 bone.refresh(using_skel, key)