Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/skeleton/meta.py: 37%
160 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-13 11:04 +0000
1import fnmatch
2import inspect
3import logging
4import os
5import string
6import sys
7import typing as t
8from deprecated.sphinx import deprecated
9from .adapter import ViurTagsSearchAdapter
10from ..bones.base import BaseBone, ReadFromClientErrorSeverity, getSystemInitialized
11from .. import db, utils
12from ..config import conf
15_UNDEFINED_KINDNAME = object()
16ABSTRACT_SKEL_CLS_SUFFIX = "AbstractSkel"
17KeyType: t.TypeAlias = db.Key | str | int
20class MetaBaseSkel(type):
21 """
22 This is the metaclass for Skeletons.
23 It is used to enforce several restrictions on bone names, etc.
24 """
25 _skelCache = {} # Mapping kindName -> SkelCls
26 _allSkelClasses = set() # list of all known skeleton classes (including Ref and Mail-Skels)
28 # List of reserved keywords and function names
29 __reserved_keywords = {
30 "all",
31 "bounce",
32 "clone",
33 "cursor",
34 "delete",
35 "errors",
36 "fromClient",
37 "fromDB",
38 "get",
39 "getCurrentSEOKeys",
40 "items",
41 "keys",
42 "limit",
43 "orderby",
44 "orderdir",
45 "patch",
46 "postDeletedHandler",
47 "postSavedHandler",
48 "preProcessBlobLocks",
49 "preProcessSerializedData",
50 "read",
51 "readonly",
52 "refresh",
53 "self",
54 "serialize",
55 "setBoneValue",
56 "structure",
57 "style",
58 "toDB",
59 "unserialize",
60 "values",
61 "write",
62 }
64 __allowed_chars = string.ascii_letters + string.digits + "_"
66 def __init__(cls, name, bases, dct, **kwargs):
67 cls.__boneMap__ = MetaBaseSkel.generate_bonemap(cls)
69 if not getSystemInitialized() and not cls.__name__.endswith(ABSTRACT_SKEL_CLS_SUFFIX): 69 ↛ 72line 69 didn't jump to line 72 because the condition on line 69 was always true
70 MetaBaseSkel._allSkelClasses.add(cls)
72 super().__init__(name, bases, dct)
74 @staticmethod
75 def generate_bonemap(cls):
76 """
77 Recursively constructs a dict of bones from
78 """
79 map = {}
81 for base in cls.__bases__:
82 if "__viurBaseSkeletonMarker__" in dir(base):
83 map |= MetaBaseSkel.generate_bonemap(base)
85 for key in cls.__dict__:
86 prop = getattr(cls, key)
88 if isinstance(prop, BaseBone):
89 if not all([c in MetaBaseSkel.__allowed_chars for c in key]): 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 raise AttributeError(f"Invalid bone name: {key!r} contains invalid characters")
91 elif key in MetaBaseSkel.__reserved_keywords: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 raise AttributeError(f"Invalid bone name: {key!r} is reserved and cannot be used")
94 map[key] = prop
96 elif prop is None and key in map: # Allow removing a bone in a subclass by setting it to None 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 del map[key]
99 return map
101 def __setattr__(self, key, value):
102 super().__setattr__(key, value)
103 if isinstance(value, BaseBone): 103 ↛ 105line 103 didn't jump to line 105 because the condition on line 103 was never true
104 # Call BaseBone.__set_name__ manually for bones that are assigned at runtime
105 value.__set_name__(self, key)
108class MetaSkel(MetaBaseSkel):
110 def __init__(cls, name, bases, dct, **kwargs):
111 super().__init__(name, bases, dct, **kwargs)
113 relNewFileName = inspect.getfile(cls) \
114 .replace(str(conf.instance.project_base_path), "") \
115 .replace(str(conf.instance.core_base_path), "")
117 # Check if we have an abstract skeleton
118 if cls.__name__.endswith(ABSTRACT_SKEL_CLS_SUFFIX): 118 ↛ 120line 118 didn't jump to line 120 because the condition on line 118 was never true
119 # Ensure that it doesn't have a kindName
120 assert cls.kindName is _UNDEFINED_KINDNAME or cls.kindName is None, \
121 "Abstract Skeletons can't have a kindName"
122 # Prevent any further processing by this class; it has to be sub-classed before it can be used
123 return
125 # Automatic determination of the kindName, if the class is not part of viur.core.
126 if ( 126 ↛ 131line 126 didn't jump to line 131 because the condition on line 126 was never true
127 cls.kindName is _UNDEFINED_KINDNAME
128 and not relNewFileName.strip(os.path.sep).startswith("viur")
129 and "viur_doc_build" not in dir(sys) # do not check during documentation build
130 ):
131 if cls.__name__.endswith("Skel"):
132 cls.kindName = cls.__name__.lower()[:-4]
133 else:
134 cls.kindName = cls.__name__.lower()
136 # Try to determine which skeleton definition takes precedence
137 if cls.kindName and cls.kindName is not _UNDEFINED_KINDNAME and cls.kindName in MetaBaseSkel._skelCache: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 relOldFileName = inspect.getfile(MetaBaseSkel._skelCache[cls.kindName]) \
139 .replace(str(conf.instance.project_base_path), "") \
140 .replace(str(conf.instance.core_base_path), "")
141 idxOld = min(
142 [x for (x, y) in enumerate(conf.skeleton_search_path) if relOldFileName.startswith(y)] + [999])
143 idxNew = min(
144 [x for (x, y) in enumerate(conf.skeleton_search_path) if relNewFileName.startswith(y)] + [999])
145 if idxNew == 999:
146 # We could not determine a priority for this class as its from a path not listed in the config
147 raise NotImplementedError(
148 "Skeletons must be defined in a folder listed in conf.skeleton_search_path")
149 elif idxOld < idxNew: # Lower index takes precedence
150 # The currently processed skeleton has a lower priority than the one we already saw - just ignore it
151 return
152 elif idxOld > idxNew:
153 # The currently processed skeleton has a higher priority, use that from now
154 MetaBaseSkel._skelCache[cls.kindName] = cls
155 else: # They seem to be from the same Package - raise as something is messed up
156 raise ValueError(f"Duplicate definition for {cls.kindName} in {relNewFileName} and {relOldFileName}")
158 # Ensure that all skeletons are defined in folders listed in conf.skeleton_search_path
159 if ( 159 ↛ 163line 159 didn't jump to line 163 because the condition on line 159 was never true
160 not any([relNewFileName.startswith(path) for path in conf.skeleton_search_path])
161 and "viur_doc_build" not in dir(sys) # do not check during documentation build
162 ):
163 raise NotImplementedError(
164 f"""{relNewFileName} must be defined in a folder listed in {conf.skeleton_search_path}""")
166 if cls.kindName and cls.kindName is not _UNDEFINED_KINDNAME: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true
167 MetaBaseSkel._skelCache[cls.kindName] = cls
169 # Auto-Add ViUR Search Tags Adapter if the skeleton has no adapter attached
170 if cls.database_adapters is _UNDEFINED_KINDNAME: 170 ↛ 174line 170 didn't jump to line 174 because the condition on line 170 was always true
171 cls.database_adapters = ViurTagsSearchAdapter()
173 # Always ensure that skel.database_adapters is an iterable
174 cls.database_adapters = utils.ensure_iterable(cls.database_adapters)
177class BaseSkeleton(object, metaclass=MetaBaseSkel):
178 """
179 This is a container-object holding information about one database entity.
181 It has to be sub-classed with individual information about the kindName of the entities
182 and its specific data attributes, the so called bones.
183 The Skeleton stores its bones in an :class:`OrderedDict`-Instance, so the definition order of the
184 contained bones remains constant.
186 :ivar key: This bone stores the current database key of this entity. \
187 Assigning to this bones value is dangerous and does *not* affect the actual key its stored in.
189 :vartype key: server.bones.BaseBone
191 :ivar creationdate: The date and time where this entity has been created.
192 :vartype creationdate: server.bones.DateBone
194 :ivar changedate: The date and time of the last change to this entity.
195 :vartype changedate: server.bones.DateBone
196 """
197 __viurBaseSkeletonMarker__ = True
198 boneMap = None
200 @classmethod
201 @deprecated(
202 version="3.7.0",
203 reason="Function renamed. Use subskel function as alternative implementation.",
204 )
205 def subSkel(cls, *subskel_names, fullClone: bool = False, **kwargs) -> "SkeletonInstance":
206 return cls.subskel(*subskel_names, clone=fullClone) # FIXME: REMOVE WITH VIUR4
208 @classmethod
209 def subskel(
210 cls,
211 *names: str,
212 bones: t.Iterable[str] = (),
213 clone: bool = False,
214 ) -> "SkeletonInstance":
215 """
216 Creates a new sub-skeleton from the current skeleton.
218 A sub-skeleton is a copy of the original skeleton, containing only a subset of its bones.
220 Sub-skeletons can either be defined using the the subSkels property of the Skeleton object,
221 or freely by giving patterns for bone names which shall be part of the sub-skeleton.
223 1. Giving names as parameter merges the bones of all Skeleton.subSkels-configurations together.
224 This is the usual behavior. By passing multiple sub-skeleton names to this function, a sub-skeleton
225 with the union of all bones of the specified sub-skeletons is returned. If an entry called "*"
226 exists in the subSkels-dictionary, the bones listed in this entry will always be part of the
227 generated sub-skeleton.
228 2. Given the *bones* parameter allows to freely specify a sub-skeleton; One specialty here is,
229 that the order of the bones can also be changed in this mode. This mode is the new way of defining
230 sub-skeletons, and might become the primary way to define sub-skeletons in future.
231 3. Both modes (1 + 2) can be combined, but then the original order of the bones is kept.
232 4. The "key" bone is automatically available in each sub-skeleton.
233 5. An fnmatch-compatible wildcard pattern is allowed both in the subSkels-bone-list and the
234 free bone list.
236 Example (TodoSkel is the example skeleton from viur-base):
237 ```py
238 # legacy mode (see 1)
239 subskel = TodoSkel.subskel("add")
240 # creates subskel: key, firstname, lastname, subject
242 # free mode (see 2) allows to specify a different order!
243 subskel = TodoSkel.subskel(bones=("subject", "message", "*stname"))
244 # creates subskel: key, subject, message, firstname, lastname
246 # mixed mode (see 3)
247 subskel = TodoSkel.subskel("add", bones=("message", ))
248 # creates subskel: key, firstname, lastname, subject, message
249 ```
251 :param bones: Allows to specify an iterator of bone names (more precisely, fnmatch-wildards) which allow
252 to freely define a subskel. If *only* this parameter is given, the order of the specification also
253 defines, the order of the list. Otherwise, the original order as defined in the skeleton is kept.
254 :param clone: If set True, performs a cloning of the used bone map, to be entirely stand-alone.
256 :return: The sub-skeleton of the specified type.
257 """
258 from_subskel = False
259 bones = list(bones)
261 for name in names:
262 # a str refers to a subskel name from the cls.subSkel dict
263 if isinstance(name, str):
264 # add bones from "*" subskel once
265 if not from_subskel:
266 bones.extend(cls.subSkels.get("*") or ())
267 from_subskel = True
269 bones.extend(cls.subSkels.get(name) or ())
271 else:
272 raise ValueError(f"Invalid subskel definition: {name!r}")
274 if from_subskel:
275 # when from_subskel is True, create bone names based on the order of the bones in the original skeleton
276 bones = tuple(k for k in cls.__boneMap__.keys() if any(fnmatch.fnmatch(k, n) for n in bones))
278 if not bones:
279 raise ValueError("The given subskel definition doesn't contain any bones!")
281 return cls(bones=bones, clone=clone)
283 @classmethod
284 def setSystemInitialized(cls):
285 for attrName in dir(cls):
286 bone = getattr(cls, attrName)
287 if isinstance(bone, BaseBone):
288 bone.setSystemInitialized()
290 @classmethod
291 def setBoneValue(
292 cls,
293 skel: "SkeletonInstance",
294 boneName: str,
295 value: t.Any,
296 append: bool = False,
297 language: t.Optional[str] = None
298 ) -> bool:
299 """
300 Allows for setting a bones value without calling fromClient or assigning a value directly.
301 Sanity-Checks are performed; if the value is invalid, that bone flips back to its original
302 (default) value and false is returned.
304 :param boneName: The name of the bone to be modified
305 :param value: The value that should be assigned. It's type depends on the type of that bone
306 :param append: If True, the given value is appended to the values of that bone instead of
307 replacing it. Only supported on bones with multiple=True
308 :param language: Language to set
310 :return: Wherever that operation succeeded or not.
311 """
312 bone = getattr(skel, boneName, None)
314 if not isinstance(bone, BaseBone):
315 raise ValueError(f"{boneName!r} is no valid bone on this skeleton ({skel!r})")
317 if language:
318 if not bone.languages:
319 raise ValueError("The bone {boneName!r} has no language setting")
320 elif language not in bone.languages:
321 raise ValueError("The language {language!r} is not available for bone {boneName!r}")
323 if value is None:
324 if append:
325 raise ValueError("Cannot append None-value to bone {boneName!r}")
327 if language:
328 skel[boneName][language] = [] if bone.multiple else None
329 else:
330 skel[boneName] = [] if bone.multiple else None
332 return True
334 _ = skel[boneName] # ensure the bone is being unserialized first
335 return bone.setBoneValue(skel, boneName, value, append, language)
337 @classmethod
338 def fromClient(
339 cls,
340 skel: "SkeletonInstance",
341 data: dict[str, list[str] | str],
342 *,
343 amend: bool = False,
344 ignore: t.Optional[t.Iterable[str]] = None,
345 ) -> bool:
346 """
347 Load supplied *data* into Skeleton.
349 This function works similar to :func:`~viur.core.skeleton.Skeleton.setValues`, except that
350 the values retrieved from *data* are checked against the bones and their validity checks.
352 Even if this function returns False, all bones are guaranteed to be in a valid state.
353 The ones which have been read correctly are set to their valid values;
354 Bones with invalid values are set back to a safe default (None in most cases).
355 So its possible to call :func:`~viur.core.skeleton.Skeleton.write` afterwards even if reading
356 data with this function failed (through this might violates the assumed consistency-model).
358 :param skel: The skeleton instance to be filled.
359 :param data: Dictionary from which the data is read.
360 :param amend: Defines whether content of data may be incomplete to amend the skel,
361 which is useful for edit-actions.
362 :param ignore: optional list of bones to be ignored; Defaults to all readonly-bones when set to None.
364 :returns: True if all data was successfully read and complete. \
365 False otherwise (e.g. some required fields where missing or where invalid).
366 """
367 complete = True
368 skel.errors = []
370 for key, bone in skel.items():
371 if (ignore is None and bone.readOnly) or key in (ignore or ()):
372 continue
374 if errors := bone.fromClient(skel, key, data):
375 for error in errors:
376 # insert current bone name into error's fieldPath
377 error.fieldPath.insert(0, str(key))
379 # logging.info(f"{key=} {error=} {skel[key]=} {bone.getEmptyValue()=}")
381 incomplete = (
382 # always when something is invalid
383 error.severity == ReadFromClientErrorSeverity.Invalid
384 or (
385 # only when path is top-level
386 len(error.fieldPath) == 1
387 and (
388 # bone is generally required
389 bool(bone.required)
390 and (
391 # and value is either empty
392 error.severity == ReadFromClientErrorSeverity.Empty
393 # or not set, depending on amending mode
394 or (
395 error.severity == ReadFromClientErrorSeverity.NotSet
396 and (amend and bone.isEmpty(skel[key]))
397 or not amend
398 )
399 )
400 )
401 )
402 )
404 # in case there are language requirements, test additionally
405 if bone.languages and isinstance(bone.required, (list, tuple)):
406 incomplete &= any([key, lang] == error.fieldPath for lang in bone.required)
408 # logging.debug(f"BaseSkel.fromClient {incomplete=} {error.severity=} {bone.required=}")
410 if incomplete:
411 complete = False
413 if conf.debug.skeleton_from_client:
414 logging.error(
415 f"""{getattr(cls, "kindName", cls.__name__)}: {".".join(error.fieldPath)}: """
416 f"""({error.severity}) {error.errorMessage}"""
417 )
418 else:
419 errors.clear()
421 skel.errors += errors
423 return complete
425 @classmethod
426 def refresh(cls, skel: "SkeletonInstance"):
427 """
428 Refresh the bones current content.
430 This function causes a refresh of all relational bones and their associated
431 information.
432 """
433 logging.debug(f"""Refreshing {skel["key"]!r} ({skel.get("name")!r})""")
435 for key, bone in skel.items():
436 if not isinstance(bone, BaseBone):
437 continue
439 _ = skel[key] # Ensure value gets loaded
440 bone.refresh(skel, key)
442 @classmethod
443 def readonly(cls, skel: "SkeletonInstance"):
444 """
445 Set all bones to readonly in the Skeleton.
446 """
447 for bone in skel.values():
448 if not isinstance(bone, BaseBone):
449 continue
450 bone.readOnly = True
452 def __new__(cls, *args, **kwargs) -> "SkeletonInstance":
453 from .instance import SkeletonInstance
454 return SkeletonInstance(cls, *args, **kwargs)