Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/prototypes/skelmodule.py: 0%
101 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-26 11:31 +0000
1import os
2import yaml
3import logging
4from viur.core import Module, db, current, errors
5from viur.core.decorators import *
6from viur.core.config import conf
7from viur.core.skeleton import skeletonByKind, Skeleton, SkeletonInstance
8import typing as t
11SINGLE_ORDER_TYPE = str | tuple[str, db.SortOrder]
12"""
13Type for exactly one sort order definitions.
14"""
16ORDER_TYPE = SINGLE_ORDER_TYPE | tuple[SINGLE_ORDER_TYPE] | list[SINGLE_ORDER_TYPE] | dict[str, str | int] | None
17"""
18Type for sort order definitions (any amount of single order definitions).
19"""
21DEFAULT_ORDER_TYPE = ORDER_TYPE | t.Callable[[db.Query], ORDER_TYPE]
22"""
23Type for default sort order definitions.
24"""
27def __load_indexes_from_file() -> dict[str, list]:
28 """
29 Loads all indexes from the index.yaml and stores it in a dictionary sorted by the module(kind)
30 :return A dictionary of indexes per module
31 """
32 indexes_dict = {}
33 try:
34 with open(os.path.join(conf.instance.project_base_path, "index.yaml"), "r") as file:
35 indexes = yaml.safe_load(file)
36 indexes = indexes.get("indexes", [])
37 for index in indexes or ():
38 index["properties"] = [_property["name"] for _property in index["properties"]]
39 indexes_dict.setdefault(index["kind"], []).append(index)
41 except FileNotFoundError:
42 logging.warning("index.yaml not found")
43 return {}
45 return indexes_dict
48DATASTORE_INDEXES = __load_indexes_from_file()
50X_VIUR_BONELIST: t.Final[str] = "X-VIUR-BONELIST"
51"""Defines the header parameter that might contain a client-defined bone list."""
54class SkelModule(Module):
55 """
56 This is the extended module prototype used by any other ViUR module prototype.
57 It a prototype which generally is bound to some database model abstracted by the ViUR skeleton system.
58 """
60 kindName: str = None
61 """
62 Name of the datastore kind that is handled by this module.
64 This information is used to bind a specific :class:`viur.core.skeleton.Skeleton`-class to this
65 prototype. By default, it is automatically determined from the module's class name, so a module named
66 `Animal` refers to a Skeleton named `AnimalSkel` and its kindName is `animal`.
68 For more information, refer to the function :func:`~_resolveSkelCls`.
69 """
71 default_order: DEFAULT_ORDER_TYPE = None
72 """
73 Allows to specify a default order for this module, which is applied when no other order is specified.
75 Setting a default_order might result in the requirement of additional indexes, which are being raised
76 and must be specified.
77 """
79 def __init__(self, *args, **kwargs):
80 super().__init__(*args, **kwargs)
82 # automatically determine kindName when not set
83 if self.kindName is None:
84 self.kindName = str(type(self).__name__).lower()
86 # assign index descriptions from index.yaml
87 self.indexes = DATASTORE_INDEXES.get(self.kindName, [])
89 def _resolveSkelCls(self, *args, **kwargs) -> t.Type[Skeleton]:
90 """
91 Retrieve the generally associated :class:`viur.core.skeleton.Skeleton` that is used by
92 the application.
94 This is either be defined by the member variable *kindName* or by a Skeleton named like the
95 application class in lower-case order.
97 If this behavior is not wanted, it can be definitely overridden by defining module-specific
98 :func:`~viewSkel`, :func:`~addSkel`, or :func:`~editSkel` functions, or by overriding this
99 function in general.
101 :return: Returns a Skeleton class that matches the application.
102 """
103 return skeletonByKind(self.kindName)
105 def baseSkel(self, *args, **kwargs) -> SkeletonInstance:
106 """
107 Returns an instance of an unmodified base skeleton for this module.
109 This function should only be used in cases where a full, unmodified skeleton of the module is required, e.g.
110 for administrative or maintenance purposes.
112 By default, baseSkel is used by :func:`~viewSkel`, :func:`~addSkel`, and :func:`~editSkel`.
113 """
114 return self.skel(**kwargs)
116 def skel(
117 self,
118 *,
119 allow_client_defined: bool = False,
120 bones: tuple[str, ...] | t.List[str] = (),
121 exclude_bones: tuple[str, ...] | t.List[str] = (),
122 **kwargs,
123 ) -> SkeletonInstance:
124 """
125 Retrieve module-specific skeleton, optionally as subskel.
127 :param allow_client_defined: Evaluates header X-VIUR-BONELIST to contain a comma-separated list of bones.
128 Using this parameter enforces that the Skeleton class has a subskel named "*" for required bones that
129 must exist.
130 :param bones: Allows to specify a list of bones to form a subskel.
131 :param exclude_bones: Provide a list of bones which are always excluded.
133 The parameters `bones` and `allow_client_defined` can be combined.
134 """
135 skel_cls = self._resolveSkelCls(**kwargs)
136 bones = set(bones) if bones else set()
138 if allow_client_defined:
139 # if bonelist := current.request.get().kwargs.get(X_VIUR_BONELIST.lower()): # DEBUG
140 if bonelist := current.request.get().request.headers.get(X_VIUR_BONELIST):
141 if "*" not in skel_cls.subSkels: # a named star-subskel "*"" must exist!
142 raise errors.BadRequest(f"Use of {X_VIUR_BONELIST!r} requires a defined star-subskel")
144 bones |= {bone.strip() for bone in bonelist.split(",")}
145 else:
146 allow_client_defined = False # is not client-defined!
148 bones.difference_update(exclude_bones)
150 # Return a subskel?
151 if bones:
152 # When coming from outside of a request, "*" is always involved.
153 if allow_client_defined:
154 current.request.get().response.vary = (X_VIUR_BONELIST, *(current.request.get().response.vary or ()))
155 return skel_cls.subskel("*", bones=bones)
157 return skel_cls(bones=bones)
159 elif exclude_bones:
160 # Return full skel, without generally excluded bones
161 bones.update(skel_cls.__boneMap__.keys())
162 bones.difference_update(exclude_bones)
163 return skel_cls(bones=bones)
165 # Otherwise, return full skeleton
166 return skel_cls()
168 def _apply_default_order(self, query: db.Query):
169 """
170 Apply the setting from `default_order` to a given db.Query.
172 The `default_order` will only be applied when the query has no other order, or is on a multquery.
173 """
175 # Apply default_order when possible!
176 if (
177 self.default_order
178 and query.queries
179 and not isinstance(query.queries, list)
180 and not query.queries.orders
181 and not current.request.get().kwargs.get("search")
182 ):
183 if callable(default_order := self.default_order):
184 default_order = default_order(query)
186 if isinstance(default_order, dict):
187 logging.debug(f"Applying filter {default_order=}")
188 query.mergeExternalFilter(default_order)
190 elif default_order:
191 logging.debug(f"Applying {default_order=}")
193 # FIXME: This ugly test can be removed when there is type that abstracts SortOrders
194 if (
195 isinstance(default_order, str)
196 or (
197 isinstance(default_order, tuple)
198 and len(default_order) == 2
199 and isinstance(default_order[0], str)
200 and isinstance(default_order[1], db.SortOrder)
201 )
202 ):
203 query.order(default_order)
204 else:
205 query.order(*default_order)
207 @force_ssl
208 @force_post
209 @exposed
210 @skey
211 @access("root")
212 def add_or_edit(self, key: db.Key | int | str, **kwargs) -> t.Any:
213 """
214 This function is intended to be used by importers.
215 Only "root"-users are allowed to use it.
216 """
218 # Adjust key
219 db_key = db.keyHelper(key, targetKind=self.kindName, adjust_kind=True)
221 # Retrieve and verify existing entry
222 db_entity = db.Get(db_key)
223 is_add = not bool(db_entity)
225 # Instanciate relevant skeleton
226 if is_add:
227 skel = self.addSkel()
228 else:
229 skel = self.editSkel()
230 skel.dbEntity = db_entity # assign existing entity
232 skel["key"] = db_key
234 if (
235 not kwargs # no data supplied
236 or not skel.fromClient(kwargs) # failure on reading into the bones
237 ):
238 # render the skeleton in the version it could as far as it could be read.
239 return self.render.render("add_or_edit", skel)
241 if is_add:
242 self.onAdd(skel)
243 else:
244 self.onEdit(skel)
246 skel.write()
248 if is_add:
249 self.onAdded(skel)
250 return self.render.addSuccess(skel)
252 self.onEdited(skel)
253 return self.render.editSuccess(skel)