Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/record.py: 58%
138 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-12 02:43 +0000
1import json
2import logging
3import typing as t
5from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity
6from viur.core import db, utils, i18n
8if t.TYPE_CHECKING: 8 ↛ 9line 8 didn't jump to line 9 because the condition on line 8 was never true
9 from ..skeleton import SkeletonInstance
12class RecordBone(BaseBone):
13 """
14 The RecordBone class is a specialized bone type used to store structured data. It inherits from
15 the BaseBone class. The RecordBone class is designed to store complex data structures, such as
16 nested dictionaries or objects, by using a related skeleton class (the using parameter) to manage
17 the internal structure of the data.
19 :param format: Optional string parameter to specify the format of the record bone.
20 :param indexed: Optional boolean parameter to indicate if the record bone is indexed.
21 Defaults to False.
22 :param using: A class that inherits from 'viur.core.skeleton.RelSkel' to be used with the
23 RecordBone.
24 :param kwargs: Additional keyword arguments to be passed to the BaseBone constructor.
25 """
26 type = "record"
28 def __init__(
29 self,
30 *,
31 format: str = None,
32 indexed: bool = False,
33 using: 'viur.core.skeleton.RelSkel' = None,
34 **kwargs
35 ):
36 from viur.core.skeleton.relskel import RelSkel
37 if not isinstance(using, type) or not issubclass(using, RelSkel):
38 raise ValueError("RecordBone requires for valid using-parameter (subclass of viur.core.skeleton.RelSkel)")
40 super().__init__(indexed=indexed, **kwargs)
41 self.using = using
42 self.format = format
43 if not format or indexed:
44 raise NotImplementedError("A RecordBone must not be indexed and must have a format set")
46 def singleValueUnserialize(self, val):
47 """
48 Unserializes a single value, creating an instance of the 'using' class and unserializing
49 the value into it.
51 :param val: The value to unserialize.
52 :return: An instance of the 'using' class with the unserialized data.
53 :raises AssertionError: If the unserialized value is not a dictionary.
54 """
55 if isinstance(val, str):
56 try:
57 value = json.loads(val)
58 except ValueError:
59 value = None
60 else:
61 value = val
63 if not value:
64 return None
66 if isinstance(value, list) and value:
67 value = value[0]
69 assert isinstance(value, dict), f"Read {value=} ({type(value)})"
71 usingSkel = self.using()
72 usingSkel.unserialize(value)
73 return usingSkel
75 def singleValueSerialize(self, value, skel: 'SkeletonInstance', name: str, parentIndexed: bool):
76 """
77 Serializes a single value by calling the serialize method of the 'using' skeleton instance.
79 :param value: The value to be serialized, which should be an instance of the 'using' skeleton.
80 :param skel: The parent skeleton instance.
81 :param name: The name of the bone.
82 :param parentIndexed: A boolean indicating if the parent bone is indexed.
83 :return: The serialized value.
84 """
85 if not value:
86 return None
88 return value.serialize(parentIndexed=False)
90 def _get_single_destinct_hash(self, value):
91 return tuple(bone._get_destinct_hash(value, name) for name, bone in self.using.__boneMap__.items())
93 def parseSubfieldsFromClient(self) -> bool:
94 """
95 Determines if the current request should attempt to parse subfields received from the client.
96 This should only be set to True if a list of dictionaries is expected to be transmitted.
97 """
98 return True
100 def singleValueFromClient(self, value, skel, bone_name, client_data):
101 usingSkel = self.using()
103 if not usingSkel.fromClient(value):
104 usingSkel.errors.append(
105 ReadFromClientError(
106 ReadFromClientErrorSeverity.Invalid,
107 i18n.translate("core.bones.error.incomplete", "Incomplete data"),
108 )
109 )
111 return usingSkel, usingSkel.errors
113 # FIXME: VIUR4 should not do "magic" anymore.
114 def performMagic(self, skel: "SkeletonInstance", name: str, isAdd: bool) -> None:
115 """
116 Forward ``performMagic`` to the sub-bones of every record value.
118 The value is only loaded when a sub-bone actually implements ``performMagic``,
119 so records without any magic are not unserialized on every write.
121 :param skel: The skeleton instance being written.
122 :param name: The name of this bone.
123 :param isAdd: Signals whether this is an add or an edit operation of the skeleton.
124 """
125 magic_bone_names = [
126 sub_bone_name for sub_bone_name, bone in self.using.__boneMap__.items()
127 if "performMagic" in vars(bone) or type(bone).performMagic is not BaseBone.performMagic
128 ]
129 if not magic_bone_names:
130 return
132 for _, _, value in self.iter_bone_value(skel, name):
133 if value is None:
134 continue
136 for sub_bone_name in magic_bone_names:
137 value.boneMap[sub_bone_name].performMagic(value, sub_bone_name, isAdd=isAdd)
139 def postSavedHandler(self, skel, boneName, key) -> None:
140 super().postSavedHandler(skel, boneName, key)
142 relational_bone_names = {
143 name for name, bone in self.using.__boneMap__.items()
144 if utils.string.is_prefix(bone.type, "relational")
145 }
146 saved_paths = set()
147 truncated = False
149 for idx, lang, value in self.iter_bone_value(skel, boneName):
150 if idx is not None and idx > 99:
151 logging.warning("postSavedHandler entry limit maximum reached")
152 truncated = True
153 break
155 if value is None:
156 continue
158 for sub_bone_name, bone in value.items():
159 path = ".".join(name for name in (boneName, lang, f"{idx or 0:02}", sub_bone_name) if name)
160 if sub_bone_name in relational_bone_names:
161 saved_paths.add(path)
163 bone.postSavedHandler(value, path, key)
165 if relational_bone_names and not truncated:
166 self._drop_stale_relations(boneName, key, relational_bone_names, saved_paths)
168 def _drop_stale_relations(
169 self,
170 bone_name: str,
171 key: db.Key,
172 relational_bone_names: set[str],
173 saved_paths: set[str],
174 ) -> None:
175 """
176 Delete the ``viur-relations`` entries of record values which no longer exist.
178 A relation is considered stale when its ``viur_src_property`` has the shape
179 ``<bone_name>[.<lang>].<index>.<sub_bone_name>`` of a known relational sub-bone,
180 but was not written by the current save (e.g. the list has been shortened or emptied).
182 Relations of other bones, of nested containers, of unknown sub-bones or unknown languages
183 are left untouched; they may belong to another version of the application.
185 :param bone_name: The name (or path, if nested) of this bone.
186 :param key: The key of the saved skeleton.
187 :param relational_bone_names: Names of the relational sub-bones of the using-skeleton.
188 :param saved_paths: The ``viur_src_property`` paths written by the current save.
189 """
190 query = db.Query("viur-relations") \
191 .filter("viur_src_kind =", key.kind) \
192 .filter("src.__key__ =", key) \
193 .filter("viur_src_property >", f"{bone_name}.") \
194 .filter("viur_src_property <", f"{bone_name}.\ufffd")
196 stale_keys = []
198 for entity in query.iter():
199 path = entity["viur_src_property"]
200 if path in saved_paths: # still in use, just written by the sub-bone's postSavedHandler
201 continue
203 # Split "<bone_name>.[<lang>.]<index>.<sub_bone_name>" behind the prefix, e.g. ["de", "01", "image"].
204 # Paths with more parts belong to a nested container, which cleans up its own relations.
205 parts = path[len(bone_name) + 1:].split(".")
206 if self.languages:
207 if len(parts) != 3 or parts[0] not in self.languages:
208 continue
209 elif len(parts) != 2:
210 continue
212 # Only drop relations this bone is responsible for; unknown sub-bones may belong to another app version
213 *_, index, sub_bone_name = parts
214 if index.isdigit() and sub_bone_name in relational_bone_names:
215 stale_keys.append(entity.key)
217 if stale_keys:
218 logging.debug(f"Delete {len(stale_keys)} stale viur-relations of {bone_name!r}")
219 db.delete(stale_keys)
221 def postDeletedHandler(self, skel, boneName, key) -> None:
222 super().postDeletedHandler(skel, boneName, key)
224 for idx, lang, value in self.iter_bone_value(skel, boneName):
225 if value is None:
226 continue
228 for sub_bone_name, bone in value.items():
229 path = ".".join(part for part in (boneName, lang, f"{idx or 0:02}", sub_bone_name) if part)
230 bone.postDeletedHandler(value, path, key)
232 def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]:
233 """
234 Collects search tags from the 'using' skeleton instance for the given bone.
236 :param skel: The parent skeleton instance.
237 :param name: The name of the bone.
238 :return: A set of search tags generated from the 'using' skeleton instance.
239 """
240 result = set()
242 for _, lang, value in self.iter_bone_value(skel, name):
243 if value is None:
244 continue
246 for key, bone in value.items():
247 if not bone.searchable:
248 continue
250 for tag in bone.getSearchTags(value, key):
251 result.add(tag)
253 return result
255 def getReferencedBlobs(self, skel: "SkeletonInstance", name: str) -> set[str]:
256 """
257 Retrieves a set of referenced blobs for the given skeleton instance and name.
259 :param skel: The skeleton instance to process.
260 :param name: The name of the bone to process.
261 :return: A set of referenced blobs.
262 """
263 result = set()
265 for _, lang, value in self.iter_bone_value(skel, name):
266 if value is None:
267 continue
269 for key, bone in value.items():
270 result |= bone.getReferencedBlobs(value, key)
272 return result
274 def getUniquePropertyIndexValues(self, skel: "SkeletonInstance", name: str) -> list[str]:
275 """
276 Returns hashes over all fields of each record value, using the serialized form as canonical input.
277 """
278 values = []
280 for _, _, using_skel in self.iter_bone_value(skel, name):
281 if using_skel is None:
282 continue
283 values.append(json.dumps(using_skel.dump(), sort_keys=True, default=str))
285 return self._hashValueForUniquePropertyIndex(values) if values else []
287 def structure(self) -> dict:
288 return super().structure() | {
289 "format": self.format,
290 "using": self.using().structure(),
291 }
293 def _atomic_dump(self, value: "SkeletonInstance") -> dict | None:
294 if value is not None:
295 return value.dump()
297 def refresh(self, skel, bone_name):
298 for _, _, using_skel in self.iter_bone_value(skel, bone_name):
299 for key, bone in using_skel.items():
300 bone.refresh(using_skel, key)
302 # When the value (acting as a skel) is marked for deletion, clear it.
303 if using_skel._cascade_deletion is True:
304 # Unset the Entity, so the skeleton becomes a False truthyness.
305 using_skel.setEntity(db.Entity())