Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/record.py: 18%
115 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-13 11:04 +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 issubclass(using, RelSkel): 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true
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: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true
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 > 99:
120 logging.warning("postSavedHandler entry limit maximum reached")
121 drop_relations_higher.clear()
122 break
124 for sub_bone_name, bone in value.items():
125 path = ".".join(name for name in (boneName, lang, f"{idx:02}", sub_bone_name) if name)
126 if utils.string.is_prefix(bone.type, "relational"):
127 drop_relations_higher[sub_bone_name] = path
129 bone.postSavedHandler(value, path, key)
131 if drop_relations_higher:
132 for viur_src_property in drop_relations_higher.values():
133 query = db.Query("viur-relations") \
134 .filter("viur_src_kind =", key.kind) \
135 .filter("src.__key__ =", key) \
136 .filter("viur_src_property >", viur_src_property)
138 logging.debug(f"Delete viur-relations with {query=}")
139 tasks.DeleteEntitiesIter.startIterOnQuery(query)
141 def postDeletedHandler(self, skel, boneName, key) -> None:
142 super().postDeletedHandler(skel, boneName, key)
144 for idx, lang, value in self.iter_bone_value(skel, boneName):
145 for sub_bone_name, bone in value.items():
146 path = ".".join(name for name in (boneName, lang, f"{idx:02}", sub_bone_name) if name)
147 bone.postDeletedHandler(value, path, key)
149 def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]:
150 """
151 Collects search tags from the 'using' skeleton instance for the given bone.
153 :param skel: The parent skeleton instance.
154 :param name: The name of the bone.
155 :return: A set of search tags generated from the 'using' skeleton instance.
156 """
157 result = set()
159 for _, lang, value in self.iter_bone_value(skel, name):
160 if value is None:
161 continue
163 for key, bone in value.items():
164 if not bone.searchable:
165 continue
167 for tag in bone.getSearchTags(value, key):
168 result.add(tag)
170 return result
172 def getSearchDocumentFields(self, valuesCache, name, prefix=""):
173 """
174 Generates a list of search document fields for the given values cache, name, and optional prefix.
176 :param dict valuesCache: A dictionary containing the cached values.
177 :param str name: The name of the bone to process.
178 :param str prefix: An optional prefix to use for the search document fields, defaults to an empty string.
179 :return: A list of search document fields.
180 :rtype: list
181 """
183 def getValues(res, skel, valuesCache, searchPrefix):
184 for key, bone in skel.items():
185 if bone.searchable:
186 res.extend(bone.getSearchDocumentFields(valuesCache, key, prefix=searchPrefix))
188 value = valuesCache.get(name)
189 res = []
191 if not value:
192 return res
193 uskel = self.using()
194 for idx, val in enumerate(value):
195 getValues(res, uskel, val, f"{prefix}{name}_{idx}")
197 return res
199 def getReferencedBlobs(self, skel: "SkeletonInstance", name: str) -> set[str]:
200 """
201 Retrieves a set of referenced blobs for the given skeleton instance and name.
203 :param skel: The skeleton instance to process.
204 :param name: The name of the bone to process.
205 :return: A set of referenced blobs.
206 """
207 result = set()
209 for _, lang, value in self.iter_bone_value(skel, name):
210 if value is None:
211 continue
213 for key, bone in value.items():
214 result |= bone.getReferencedBlobs(value, key)
216 return result
218 def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> list[str]:
219 """
220 This method is intentionally not implemented as it's not possible to determine how to derive
221 a key from the related skeleton being used (i.e., which fields to include and how).
223 """
224 raise NotImplementedError()
226 def structure(self) -> dict:
227 return super().structure() | {
228 "format": self.format,
229 "using": self.using().structure(),
230 }
232 def _atomic_dump(self, value: "SkeletonInstance") -> dict | None:
233 if value is not None:
234 return value.dump()
236 def refresh(self, skel, bone_name):
237 for _, _, using_skel in self.iter_bone_value(skel, bone_name):
238 for key, bone in using_skel.items():
239 bone.refresh(using_skel, key)
241 # When the value (acting as a skel) is marked for deletion, clear it.
242 if using_skel._cascade_deletion is True:
243 # Unset the Entity, so the skeleton becomes a False truthyness.
244 using_skel.setEntity(db.Entity())