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

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 {value=} ({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 

101 if not usingSkel.fromClient(value): 

102 usingSkel.errors.append( 

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

104 ) 

105 

106 return usingSkel, usingSkel.errors 

107 

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

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

110 

111 for _, lang, value in self.iter_bone_value(skel, boneName): 

112 for bone_name, bone in value.items(): 

113 bone.postSavedHandler(value, bone_name, None) 

114 

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

116 """ 

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

118 

119 :param skel: The parent skeleton instance. 

120 :param name: The name of the bone. 

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

122 """ 

123 result = set() 

124 

125 for _, lang, value in self.iter_bone_value(skel, name): 

126 if value is None: 

127 continue 

128 

129 for key, bone in value.items(): 

130 if not bone.searchable: 

131 continue 

132 

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

134 result.add(tag) 

135 

136 return result 

137 

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

139 """ 

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

141 

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

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

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

145 :return: A list of search document fields. 

146 :rtype: list 

147 """ 

148 

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

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

151 if bone.searchable: 

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

153 

154 value = valuesCache.get(name) 

155 res = [] 

156 

157 if not value: 

158 return res 

159 uskel = self.using() 

160 for idx, val in enumerate(value): 

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

162 

163 return res 

164 

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

166 """ 

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

168 

169 :param skel: The skeleton instance to process. 

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

171 :return: A set of referenced blobs. 

172 """ 

173 result = set() 

174 

175 for _, lang, value in self.iter_bone_value(skel, name): 

176 if value is None: 

177 continue 

178 

179 for key, bone in value.items(): 

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

181 

182 return result 

183 

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

185 """ 

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

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

188 

189 """ 

190 raise NotImplementedError() 

191 

192 def structure(self) -> dict: 

193 return super().structure() | { 

194 "format": self.format, 

195 "using": self.using().structure(), 

196 } 

197 

198 def refresh(self, skel, bone_name): 

199 for _, lang, value in self.iter_bone_value(skel, bone_name): 

200 if value is None: 

201 continue 

202 

203 for key, bone in value.items(): 

204 bone.refresh(value, key)