Coverage for  / home / runner / work / viur-core / viur-core / viur / src / viur / core / skeleton / meta.py: 37%

159 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-02-25 14:23 +0000

1import fnmatch 

2import inspect 

3import logging # noqa 

4import os 

5import string 

6import sys 

7import typing as t 

8 

9from deprecated.sphinx import deprecated 

10 

11from .adapter import ViurTagsSearchAdapter 

12from .. import db, utils 

13from ..bones.base import BaseBone, ReadFromClientErrorSeverity, getSystemInitialized 

14from ..config import conf 

15 

16_UNDEFINED_KINDNAME = object() 

17ABSTRACT_SKEL_CLS_SUFFIX = "AbstractSkel" 

18 

19 

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) 

27 

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 } 

63 

64 __allowed_chars = string.ascii_letters + string.digits + "_" 

65 

66 def __init__(cls, name, bases, dct, **kwargs): 

67 cls.__boneMap__ = MetaBaseSkel.generate_bonemap(cls) 

68 

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) 

71 

72 super().__init__(name, bases, dct) 

73 

74 @staticmethod 

75 def generate_bonemap(cls): 

76 """ 

77 Recursively constructs a dict of bones from 

78 """ 

79 map = {} 

80 

81 for base in cls.__bases__: 

82 if "__viurBaseSkeletonMarker__" in dir(base): 

83 map |= MetaBaseSkel.generate_bonemap(base) 

84 

85 for key in cls.__dict__: 

86 prop = getattr(cls, key) 

87 

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") 

93 

94 map[key] = prop 

95 

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] 

98 

99 return map 

100 

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) 

106 

107 

108class MetaSkel(MetaBaseSkel): 

109 

110 def __init__(cls, name, bases, dct, **kwargs): 

111 super().__init__(name, bases, dct, **kwargs) 

112 

113 relNewFileName = inspect.getfile(cls) \ 

114 .replace(str(conf.instance.project_base_path), "") \ 

115 .replace(str(conf.instance.core_base_path), "") 

116 

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 

124 

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() 

135 

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}") 

157 

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}""") 

165 

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 

168 

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() 

172 

173 # Always ensure that skel.database_adapters is an iterable 

174 cls.database_adapters = utils.ensure_iterable(cls.database_adapters) 

175 

176 

177# FIXME: Why is this in meta if this isn't a metaclass? it belongs to skeleton or a own module! 

178class BaseSkeleton(object, metaclass=MetaBaseSkel): 

179 """ 

180 This is a container-object holding information about one database entity. 

181 

182 It has to be sub-classed with individual information about the kindName of the entities 

183 and its specific data attributes, the so called bones. 

184 The Skeleton stores its bones in an :class:`OrderedDict`-Instance, so the definition order of the 

185 contained bones remains constant. 

186 

187 :ivar key: This bone stores the current database key of this entity. \ 

188 Assigning to this bones value is dangerous and does *not* affect the actual key its stored in. 

189 

190 :vartype key: server.bones.BaseBone 

191 

192 :ivar creationdate: The date and time where this entity has been created. 

193 :vartype creationdate: server.bones.DateBone 

194 

195 :ivar changedate: The date and time of the last change to this entity. 

196 :vartype changedate: server.bones.DateBone 

197 """ 

198 __viurBaseSkeletonMarker__ = True 

199 boneMap = None 

200 

201 @classmethod 

202 @deprecated( 

203 version="3.7.0", 

204 reason="Function renamed. Use subskel function as alternative implementation.", 

205 ) 

206 def subSkel(cls, *subskel_names, fullClone: bool = False, **kwargs) -> "SkeletonInstance": 

207 return cls.subskel(*subskel_names, clone=fullClone) # FIXME: REMOVE WITH VIUR4 

208 

209 @classmethod 

210 def subskel( 

211 cls, 

212 *names: str, 

213 bones: t.Iterable[str] = (), 

214 clone: bool = False, 

215 ) -> "SkeletonInstance": 

216 """ 

217 Creates a new sub-skeleton from the current skeleton. 

218 

219 A sub-skeleton is a copy of the original skeleton, containing only a subset of its bones. 

220 

221 Sub-skeletons can either be defined using the the subSkels property of the Skeleton object, 

222 or freely by giving patterns for bone names which shall be part of the sub-skeleton. 

223 

224 1. Giving names as parameter merges the bones of all Skeleton.subSkels-configurations together. 

225 This is the usual behavior. By passing multiple sub-skeleton names to this function, a sub-skeleton 

226 with the union of all bones of the specified sub-skeletons is returned. If an entry called "*" 

227 exists in the subSkels-dictionary, the bones listed in this entry will always be part of the 

228 generated sub-skeleton. 

229 2. Given the *bones* parameter allows to freely specify a sub-skeleton; One specialty here is, 

230 that the order of the bones can also be changed in this mode. This mode is the new way of defining 

231 sub-skeletons, and might become the primary way to define sub-skeletons in future. 

232 3. Both modes (1 + 2) can be combined, but then the original order of the bones is kept. 

233 4. The "key" bone is automatically available in each sub-skeleton. 

234 5. An fnmatch-compatible wildcard pattern is allowed both in the subSkels-bone-list and the 

235 free bone list. 

236 

237 Example (TodoSkel is the example skeleton from viur-base): 

238 ```py 

239 # legacy mode (see 1) 

240 subskel = TodoSkel.subskel("add") 

241 # creates subskel: key, firstname, lastname, subject 

242 

243 # free mode (see 2) allows to specify a different order! 

244 subskel = TodoSkel.subskel(bones=("subject", "message", "*stname")) 

245 # creates subskel: key, subject, message, firstname, lastname 

246 

247 # mixed mode (see 3) 

248 subskel = TodoSkel.subskel("add", bones=("message", )) 

249 # creates subskel: key, firstname, lastname, subject, message 

250 ``` 

251 

252 :param bones: Allows to specify an iterator of bone names (more precisely, fnmatch-wildards) which allow 

253 to freely define a subskel. If *only* this parameter is given, the order of the specification also 

254 defines, the order of the list. Otherwise, the original order as defined in the skeleton is kept. 

255 :param clone: If set True, performs a cloning of the used bone map, to be entirely stand-alone. 

256 

257 :return: The sub-skeleton of the specified type. 

258 """ 

259 from_subskel = False 

260 bones = list(bones) 

261 

262 for name in names: 

263 # a str refers to a subskel name from the cls.subSkel dict 

264 if isinstance(name, str): 

265 # add bones from "*" subskel once 

266 if not from_subskel: 

267 bones.extend(cls.subSkels.get("*") or ()) 

268 from_subskel = True 

269 

270 bones.extend(cls.subSkels.get(name) or ()) 

271 

272 else: 

273 raise ValueError(f"Invalid subskel definition: {name!r}") 

274 

275 if from_subskel: 

276 # when from_subskel is True, create bone names based on the order of the bones in the original skeleton 

277 bones = tuple(k for k in cls.__boneMap__.keys() if any(fnmatch.fnmatch(k, n) for n in bones)) 

278 

279 if not bones: 

280 raise ValueError("The given subskel definition doesn't contain any bones!") 

281 

282 return cls(bones=bones, clone=clone) 

283 

284 @classmethod 

285 def setSystemInitialized(cls): 

286 for attrName in dir(cls): 

287 bone = getattr(cls, attrName) 

288 if isinstance(bone, BaseBone): 

289 bone.setSystemInitialized() 

290 

291 @classmethod 

292 def setBoneValue( 

293 cls, 

294 skel: "SkeletonInstance", 

295 boneName: str, 

296 value: t.Any, 

297 append: bool = False, 

298 language: t.Optional[str] = None 

299 ) -> bool: 

300 """ 

301 Allows for setting a bones value without calling fromClient or assigning a value directly. 

302 Sanity-Checks are performed; if the value is invalid, that bone flips back to its original 

303 (default) value and false is returned. 

304 

305 :param boneName: The name of the bone to be modified 

306 :param value: The value that should be assigned. It's type depends on the type of that bone 

307 :param append: If True, the given value is appended to the values of that bone instead of 

308 replacing it. Only supported on bones with multiple=True 

309 :param language: Language to set 

310 

311 :return: Wherever that operation succeeded or not. 

312 """ 

313 bone = getattr(skel, boneName, None) 

314 

315 if not isinstance(bone, BaseBone): 

316 raise ValueError(f"{boneName!r} is no valid bone on this skeleton ({skel!r})") 

317 

318 if language: 

319 if not bone.languages: 

320 raise ValueError("The bone {boneName!r} has no language setting") 

321 elif language not in bone.languages: 

322 raise ValueError("The language {language!r} is not available for bone {boneName!r}") 

323 

324 if value is None: 

325 if append: 

326 raise ValueError("Cannot append None-value to bone {boneName!r}") 

327 

328 if language: 

329 skel[boneName][language] = [] if bone.multiple else None 

330 else: 

331 skel[boneName] = [] if bone.multiple else None 

332 

333 return True 

334 

335 _ = skel[boneName] # ensure the bone is being unserialized first 

336 return bone.setBoneValue(skel, boneName, value, append, language) 

337 

338 @classmethod 

339 def fromClient( 

340 cls, 

341 skel: "SkeletonInstance", 

342 data: dict[str, list[str] | str], 

343 *, 

344 amend: bool = False, 

345 ignore: t.Optional[t.Iterable[str]] = None, 

346 ) -> bool: 

347 """ 

348 Load supplied *data* into Skeleton. 

349 

350 This function works similar to :func:`~viur.core.skeleton.Skeleton.setValues`, except that 

351 the values retrieved from *data* are checked against the bones and their validity checks. 

352 

353 Even if this function returns False, all bones are guaranteed to be in a valid state. 

354 The ones which have been read correctly are set to their valid values; 

355 Bones with invalid values are set back to a safe default (None in most cases). 

356 So its possible to call :func:`~viur.core.skeleton.Skeleton.write` afterwards even if reading 

357 data with this function failed (through this might violates the assumed consistency-model). 

358 

359 :param skel: The skeleton instance to be filled. 

360 :param data: Dictionary from which the data is read. 

361 :param amend: Defines whether content of data may be incomplete to amend the skel, 

362 which is useful for edit-actions. 

363 :param ignore: optional list of bones to be ignored; Defaults to all readonly-bones when set to None. 

364 

365 :returns: True if all data was successfully read and complete. \ 

366 False otherwise (e.g. some required fields where missing or where invalid). 

367 """ 

368 complete = True 

369 skel.errors = [] 

370 

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

372 if (ignore is None and bone.readOnly) or key in (ignore or ()): 

373 continue 

374 

375 if errors := bone.fromClient(skel, key, data): 

376 for error in errors: 

377 # insert current bone name into error's fieldPath 

378 error.fieldPath.insert(0, str(key)) 

379 

380 # logging.info(f"{key=} {error=} {skel[key]=} {bone.getEmptyValue()=}") 

381 

382 incomplete = ( 

383 # always when something is invalid 

384 error.severity == ReadFromClientErrorSeverity.Invalid 

385 or ( 

386 # only when path is top-level 

387 len(error.fieldPath) == 1 

388 and ( 

389 # bone is generally required 

390 bool(bone.required) 

391 and ( 

392 # and value is either empty 

393 error.severity == ReadFromClientErrorSeverity.Empty 

394 # or not set, depending on amending mode 

395 or ( 

396 error.severity == ReadFromClientErrorSeverity.NotSet 

397 and (amend and bone.isEmpty(skel[key])) 

398 or not amend 

399 ) 

400 ) 

401 ) 

402 ) 

403 ) 

404 

405 # in case there are language requirements, test additionally 

406 if bone.languages and isinstance(bone.required, (list, tuple)): 

407 incomplete &= any([key, lang] == error.fieldPath for lang in bone.required) 

408 

409 # logging.debug(f"BaseSkel.fromClient {incomplete=} {error.severity=} {bone.required=}") 

410 

411 if incomplete: 

412 complete = False 

413 

414 if conf.debug.skeleton_from_client: 

415 logging.error( 

416 f"""{getattr(cls, "kindName", cls.__name__)}: {".".join(error.fieldPath)}: """ 

417 f"""({error.severity}) {error.errorMessage}""" 

418 ) 

419 else: 

420 errors.clear() 

421 

422 skel.errors += errors 

423 

424 return complete 

425 

426 @classmethod 

427 def refresh(cls, skel: "SkeletonInstance"): 

428 """ 

429 Refresh the bones current content. 

430 

431 This function causes a refresh of all relational bones and their associated 

432 information. 

433 """ 

434 logging.debug(f"""Refreshing {skel["key"]!r} ({skel.get("name")!r})""") 

435 

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

437 if not isinstance(bone, BaseBone): 

438 continue 

439 

440 _ = skel[key] # Ensure value gets loaded 

441 bone.refresh(skel, key) 

442 

443 @classmethod 

444 def readonly(cls, skel: "SkeletonInstance"): 

445 """ 

446 Set all bones to readonly in the Skeleton. 

447 """ 

448 for bone in skel.values(): 

449 if not isinstance(bone, BaseBone): 

450 continue 

451 bone.readOnly = True 

452 

453 def __new__(cls, *args, **kwargs) -> "SkeletonInstance": 

454 from .instance import SkeletonInstance 

455 return SkeletonInstance(cls, *args, **kwargs)