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

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 getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: 

107 """ 

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

109 

110 :param skel: The parent skeleton instance. 

111 :param name: The name of the bone. 

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

113 """ 

114 result = set() 

115 

116 using_skel_cache = self.using() 

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

118 if value is None: 

119 continue 

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

121 if not bone.searchable: 

122 continue 

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

124 result.add(tag) 

125 

126 return result 

127 

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

129 """ 

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

131 

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

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

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

135 :return: A list of search document fields. 

136 :rtype: list 

137 """ 

138 

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

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

141 if bone.searchable: 

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

143 

144 value = valuesCache.get(name) 

145 res = [] 

146 

147 if not value: 

148 return res 

149 uskel = self.using() 

150 for idx, val in enumerate(value): 

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

152 

153 return res 

154 

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

156 """ 

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

158 

159 :param skel: The skeleton instance to process. 

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

161 :return: A set of referenced blobs. 

162 """ 

163 result = set() 

164 

165 using_skel_cache = self.using() 

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

167 if value is None: 

168 continue 

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

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

171 

172 return result 

173 

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

175 """ 

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

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

178 

179 """ 

180 raise NotImplementedError() 

181 

182 def structure(self) -> dict: 

183 return super().structure() | { 

184 "format": self.format, 

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