Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/record.py: 39%
111 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 15:02 +0000
1import json
2import logging
3import typing as t
5from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity
6from viur.core import db, utils, tasks, 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 def postSavedHandler(self, skel, boneName, key) -> None:
114 super().postSavedHandler(skel, boneName, key)
116 drop_relations_higher = {}
118 for idx, lang, value in self.iter_bone_value(skel, boneName):
119 if idx is not None and idx > 99: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 logging.warning("postSavedHandler entry limit maximum reached")
121 drop_relations_higher.clear()
122 break
124 if value is None:
125 continue
127 for sub_bone_name, bone in value.items():
128 path = ".".join(name for name in (boneName, lang, f"{idx or 0:02}", sub_bone_name) if name)
129 if utils.string.is_prefix(bone.type, "relational"): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 drop_relations_higher[sub_bone_name] = path
132 bone.postSavedHandler(value, path, key)
134 if drop_relations_higher: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 for viur_src_property in drop_relations_higher.values():
136 query = db.Query("viur-relations") \
137 .filter("viur_src_kind =", key.kind) \
138 .filter("src.__key__ =", key) \
139 .filter("viur_src_property >", viur_src_property)
141 logging.debug(f"Delete viur-relations with {query=}")
142 tasks.DeleteEntitiesIter.startIterOnQuery(query)
144 def postDeletedHandler(self, skel, boneName, key) -> None:
145 super().postDeletedHandler(skel, boneName, key)
147 for idx, lang, value in self.iter_bone_value(skel, boneName):
148 if value is None:
149 continue
151 for sub_bone_name, bone in value.items():
152 path = ".".join(part for part in (boneName, lang, f"{idx or 0:02}", sub_bone_name) if part)
153 bone.postDeletedHandler(value, path, key)
155 def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]:
156 """
157 Collects search tags from the 'using' skeleton instance for the given bone.
159 :param skel: The parent skeleton instance.
160 :param name: The name of the bone.
161 :return: A set of search tags generated from the 'using' skeleton instance.
162 """
163 result = set()
165 for _, lang, value in self.iter_bone_value(skel, name):
166 if value is None:
167 continue
169 for key, bone in value.items():
170 if not bone.searchable:
171 continue
173 for tag in bone.getSearchTags(value, key):
174 result.add(tag)
176 return result
178 def getReferencedBlobs(self, skel: "SkeletonInstance", name: str) -> set[str]:
179 """
180 Retrieves a set of referenced blobs for the given skeleton instance and name.
182 :param skel: The skeleton instance to process.
183 :param name: The name of the bone to process.
184 :return: A set of referenced blobs.
185 """
186 result = set()
188 for _, lang, value in self.iter_bone_value(skel, name):
189 if value is None:
190 continue
192 for key, bone in value.items():
193 result |= bone.getReferencedBlobs(value, key)
195 return result
197 def getUniquePropertyIndexValues(self, skel: "SkeletonInstance", name: str) -> list[str]:
198 """
199 Returns hashes over all fields of each record value, using the serialized form as canonical input.
200 """
201 values = []
203 for _, _, using_skel in self.iter_bone_value(skel, name):
204 if using_skel is None:
205 continue
206 values.append(json.dumps(using_skel.dump(), sort_keys=True, default=str))
208 return self._hashValueForUniquePropertyIndex(values) if values else []
210 def structure(self) -> dict:
211 return super().structure() | {
212 "format": self.format,
213 "using": self.using().structure(),
214 }
216 def _atomic_dump(self, value: "SkeletonInstance") -> dict | None:
217 if value is not None:
218 return value.dump()
220 def refresh(self, skel, bone_name):
221 for _, _, using_skel in self.iter_bone_value(skel, bone_name):
222 for key, bone in using_skel.items():
223 bone.refresh(using_skel, key)
225 # When the value (acting as a skel) is marked for deletion, clear it.
226 if using_skel._cascade_deletion is True:
227 # Unset the Entity, so the skeleton becomes a False truthyness.
228 using_skel.setEntity(db.Entity())