Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/module.py: 73%
213 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-12 02:43 +0000
1import copy
2import enum
3import functools
4import inspect
5import types
6import typing as t
7import logging
8from viur.core import db, errors, current, utils
9from viur.core.config import conf
12class Method:
13 """
14 Abstraction wrapper for any public available method.
15 """
17 @classmethod
18 def ensure(cls, func: t.Callable | "Method") -> "Method":
19 """
20 Ensures the provided `func` parameter is either a Method already, or turns it
21 into a Method. This is done to avoid stacking Method objects, which may create
22 unwanted results.
23 """
24 if isinstance(func, Method):
25 return func
27 return cls(func)
29 def __init__(self, func: t.Callable):
30 # Content
31 self._func = func
32 self.__name__ = func.__name__
33 self._instance = None
35 # Attributes
36 self.exposed = None # None = unexposed, True = exposed, False = internal exposed
37 self.ssl = False
38 self.methods = ("GET", "POST", "HEAD", "OPTIONS")
39 self.seo_language_map = None
40 self.cors_allow_headers = None
41 self.additional_descr = {}
42 self.skey = None
44 # Inspection
45 self.signature = inspect.signature(self._func)
47 # Guards
48 self.guards = []
50 def __get__(self, obj, objtype=None):
51 """
52 This binds the Method to an object.
54 To do it, the Method instance is copied and equipped with the individual _instance member.
55 """
56 if obj:
57 bound = copy.copy(self)
58 bound._instance = obj
59 return bound
61 return self
63 def __call__(self, *args, **kwargs):
64 """
65 Calls the method with given args and kwargs.
67 Prepares and filters argument values from args and kwargs regarding self._func's signature and type annotations,
68 if present.
70 Method objects normally wrap functions which are externally exposed. Therefore, any arguments passed from the
71 client are str-values, and are automatically parsed when equipped with type-annotations.
73 This preparation of arguments therefore inspects the target function as follows
74 - incoming values are parsed to their particular type, if type annotations are present
75 - parameters in *args and **kwargs are being checked against their signature; only relevant values are being
76 passed, anything else is thrown away.
77 - execution of guard configurations from @skey and @access, if present
78 """
80 if trace := conf.debug.trace: 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 logging.debug(f"calling {self._func=} with raw {args=}, {kwargs=}")
83 def parse_value_by_annotation(annotation: type, name: str, value: str | list | tuple) -> t.Any:
84 """
85 Tries to parse a value according to a given type.
86 May be called recursively to handle unions, lists and tuples as well.
87 """
88 # logging.debug(f"{annotation=} | {name=} | {value=}")
90 # simple types
91 if annotation is str:
92 return str(value)
93 elif annotation is int:
94 return int(value)
95 elif annotation is float:
96 return float(value)
97 elif annotation is bool:
98 return utils.parse.bool(value)
99 elif annotation is types.NoneType or annotation is None:
100 if value in (None, "None", "null"): 100 ↛ 102line 100 didn't jump to line 102 because the condition on line 100 was always true
101 return None
102 raise ValueError(f"Expected None for parameter {name}. Got: {value!r}")
104 # complex types
105 origin_type = t.get_origin(annotation)
107 if origin_type is list and len(annotation.__args__) == 1:
108 if not isinstance(value, list):
109 value = [value]
111 return [parse_value_by_annotation(annotation.__args__[0], name, item) for item in value]
113 elif origin_type is tuple and len(annotation.__args__) == 1:
114 if not isinstance(value, tuple):
115 value = (value, )
117 return tuple(parse_value_by_annotation(annotation.__args__[0], name, item) for item in value)
119 elif origin_type is t.Literal:
120 if not any(value == str(literal) for literal in annotation.__args__):
121 raise errors.NotAcceptable(f"Expecting any of {annotation.__args__} for {name}")
123 return value
125 elif origin_type is t.Union or isinstance(annotation, types.UnionType):
126 for i, sub_annotation in enumerate(annotation.__args__): 126 ↛ 156line 126 didn't jump to line 156 because the loop on line 126 didn't complete
127 try:
128 return parse_value_by_annotation(sub_annotation, name, value)
129 # Literal, Enum and unhandled members reject a value with NotAcceptable instead of a
130 # ValueError; try the next member anyway
131 except (ValueError, errors.NotAcceptable):
132 if i == len(annotation.__args__) - 1:
133 raise
135 elif annotation is db.Key:
136 if isinstance(value, db.Key):
137 return value
139 elif isinstance(value, str): # Maybe we have an url encoded Key 139 ↛ 145line 139 didn't jump to line 145 because the condition on line 139 was always true
140 try:
141 return db.normalize_key(value)
142 except Exception:
143 pass
145 return parse_value_by_annotation(int | str, name, value)
147 elif isinstance(annotation, enum.EnumMeta):
148 try:
149 return annotation(value)
150 except ValueError as exc:
151 for value_, member in annotation._value2member_map_.items():
152 if str(value) == str(value_): # Do a string comparison, it could be a IntEnum
153 return member
154 raise errors.NotAcceptable(f"{' '.join(exc.args)} for {name}") from exc
156 raise errors.NotAcceptable(f"Unhandled type {annotation=} for {name}={value!r}")
158 # examine parameters
159 args_iter = iter(args)
161 parsed_args = []
162 parsed_kwargs = {}
163 varargs = []
164 varkwargs = False
166 for i, (param_name, param) in enumerate(self.signature.parameters.items()):
167 if self._instance and i == 0 and param_name == "self":
168 continue
170 param_type = param.annotation
171 param_required = param.default is param.empty
173 # take positional parameters first
174 if param.kind in (
175 inspect.Parameter.POSITIONAL_OR_KEYWORD,
176 inspect.Parameter.POSITIONAL_ONLY
177 ):
178 try:
179 value = next(args_iter)
181 if param_type is not param.empty:
182 value = parse_value_by_annotation(param_type, param_name, value)
184 parsed_args.append(value)
185 continue
186 except StopIteration:
187 pass
189 # otherwise take kwargs or variadics
190 if (
191 param.kind in (
192 inspect.Parameter.POSITIONAL_OR_KEYWORD,
193 inspect.Parameter.KEYWORD_ONLY
194 )
195 and param_name in kwargs
196 ):
197 value = kwargs.pop(param_name)
199 if param_type is not param.empty:
200 try:
201 value = parse_value_by_annotation(param_type, param_name, value)
202 except ValueError as exc:
203 raise errors.NotAcceptable(f"Invalid value for {param_name}") from exc
205 parsed_kwargs[param_name] = value
207 elif param.kind == inspect.Parameter.VAR_POSITIONAL:
208 varargs = list(args_iter)
209 elif param.kind == inspect.Parameter.VAR_KEYWORD:
210 varkwargs = True
211 elif param_required:
212 if self.skey and param_name == self.skey["forward_payload"]: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 continue
215 raise errors.NotAcceptable(f"Missing required parameter {param_name!r}")
217 # Here's a short clarification on the variables used here:
218 #
219 # - parsed_args = tuple of (the type-parsed) arguments that have been assigned based on the signature
220 # - parsed_kwargs = dict of (the type-parsed) keyword arguments that have been assigned based on the signature
221 # - args = either parsed_args, or parsed_args + remaining args if the function accepts *args
222 # - kwargs = either parsed_kwars, or parsed_kwargs | remaining kwargs if the function accepts **kwargs
223 # - varargs = indicator that the args also contain variable args (*args)
224 # - varkwargs = indicator that variable kwargs (**kwargs) are also contained in the kwargs
225 #
227 # Extend args to any varargs, and redefine args
228 args = tuple(parsed_args + varargs)
230 # always take "skey"-parameter name, when configured, as parsed_kwargs
231 if self.skey and self.skey["name"] in kwargs: 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true
232 parsed_kwargs[self.skey["name"]] = kwargs.pop(self.skey["name"])
234 # When varkwargs are accepted, merge parsed_kwargs and kwargs, otherwise just use parsed_kwargs
235 if varkwargs := varkwargs and bool(kwargs):
236 kwargs = parsed_kwargs | kwargs
237 else:
238 kwargs = parsed_kwargs
240 # Trace message for final call configuration
241 if conf.debug.trace: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 logging.debug(f"calling {self._func=} with cleaned {args=}, {kwargs=}")
243 # call decorators in reversed because they are added in the reversed order
244 for func in reversed(self.guards):
245 func(args=args, kwargs=kwargs, varargs=varargs, varkwargs=varkwargs)
247 # call with instance when provided
248 if self._instance:
249 return self._func(self._instance, *args, **kwargs)
251 return self._func(*args, **kwargs)
253 def describe(self) -> dict:
254 """
255 Describes the Method with a
256 """
257 return_doc = t.get_type_hints(self._func).get("return")
259 return {
260 "args": {
261 param.name: {
262 "type": str(param.annotation) if param.annotation is not inspect.Parameter.empty else None,
263 "default": str(param.default) if param.default is not inspect.Parameter.empty else None,
264 }
265 for param in self.signature.parameters.values()
266 },
267 "returns": str(return_doc).strip() if return_doc else None,
268 "accepts": self.methods,
269 "docs": self._func.__doc__.strip() if self._func.__doc__ else None,
270 "aliases": tuple(self.seo_language_map.keys()) if self.seo_language_map else None,
271 } | self.additional_descr
273 def register(self, target: dict, name: str, language: str | None = None):
274 """
275 Registers the Method under `name` and eventually some customized SEO-name for the provided language
276 """
277 if self.exposed is None:
278 return
280 target[name] = self
282 # reassign for SEO mapping as well
283 if self.seo_language_map:
284 for lang in tuple(self.seo_language_map.keys()) if not language else (language, ):
285 if translated_name := self.seo_language_map.get(lang):
286 target[translated_name] = self
289class Module:
290 """
291 This is the root module prototype that serves a minimal module in the ViUR system without any other bindings.
292 """
294 handler: str | t.Callable = None
295 """
296 This is the module's handler, respectively its type.
297 Use the @property-decorator in specific Modules to construct the handler's value dynamically.
298 A module without a handler setting cannot be described, so cannot be handled by admin-tools.
299 """
301 accessRights: tuple[str] = None
302 """
303 If set, a tuple of access rights (like add, edit, delete) that this module supports.
305 These will be prefixed on instance startup with the actual module name (becoming file-add, file-edit etc)
306 and registered in ``conf.user.access_rights`` so these will be available on the access bone in user/add
307 or user/edit.
308 """
310 roles: dict = {}
311 r"""
312 Allows to specify role settings for a module.
314 Defaults to no role definition, which ignores the module entirely in the role-system.
315 In this case, access rights can still be set individually on the user's access bone.
317 A "*" wildcard can either be used as key or as value to allow for "all roles", or "all rights".
319 .. code-block:: python
321 # Example
322 roles = {
323 "*": "view", # Any role may only "view"
324 "editor": ("add", "edit"), # Role "editor" may "add" or "edit", but not "delete"
325 "admin": "*", # Role "admin" can do everything
326 }
328 """
330 seo_language_map: dict[str: str] = {}
331 r"""
332 The module name is the first part of a URL.
333 SEO-identifiers have to be set as class-attribute ``seo_language_map`` of type ``dict[str, str]`` in the module.
334 It maps a *language* to the according *identifier*.
336 .. code-block:: python
337 :name: module seo-map
338 :caption: modules/myorders.py
339 :emphasize-lines: 4-7
341 from viur.core.prototypes import List
343 class MyOrders(List):
344 seo_language_map = {
345 "de": "bestellungen",
346 "en": "orders",
347 }
349 By default the module would be available under */myorders*, the lowercase module name.
350 With the defined :attr:`seo_language_map`, it will become available as */de/bestellungen* and */en/orders*.
352 Great, this part is now user and robot friendly :)
353 """
355 adminInfo: dict[str, t.Any] | t.Callable = None
356 """
357 This is a ``dict`` holding the information necessary for the Vi/Admin to handle this module.
359 name: ``str``
360 Human-readable module name that will be shown in the admin tool.
362 handler: ``str`` (``list``, ``tree`` or ``singleton``):
363 Allows to override the handler provided by the module. Set this only when *really* necessary,
364 otherwise it can be left out and is automatically injected by the Module's prototype.
366 icon: ``str``
367 (Optional) Either the Shoelace icon library name or a path relative to the project's deploy folder
368 (e.g. /static/icons/viur.svg) for the icon used in the admin tool for this module.
370 columns: ``List[str]``
371 (Optional) List of columns (bone names) that are displayed by default.
372 Used only by the List handler.
374 filter: ``Dict[str, str]``
375 (Optional) Dictionary of additional parameters that will be send along when
376 fetching entities from the server. Can be used to filter the entities being displayed on the
377 client-side.
379 display: ``str`` ("default", "hidden" or "group")
380 (Optional) "hidden" will hide the module in the admin tool's main bar.
381 (itwill not be accessible directly, however it's registered with the frontend so it can be used in a
382 relational bone). "group" will show this module in the main bar, but it will not be clickable.
383 Clicking it will just try to expand it (assuming there are additional views defined).
385 preview: ``Union[str, Dict[str, str]]``
386 (Optional) A url that will be opened in a new tab and is expected to display
387 the entity selected in the table. Can be “/{{module}}/view/{{key}}", with {{module}} and {{key}} getting
388 replaced as needed. If more than one preview-url is needed, supply a dictionary where the key is
389 the URL and the value the description shown to the user.
391 views: ``List[Dict[str, t.Any]]``
392 (Optional) List of nested adminInfo like dictionaries. Used to define
393 additional views on the module. Useful f.e. for an order module, where you want separate list of
394 "payed orders", "unpayed orders", "orders waiting for shipment", etc. If such views are defined,
395 the top-level entry in the menu bar will expand if clicked, revealing these additional filters.
397 actions: ``List[str]``
398 (Optional) List of actions supported by this modules. Actions can be defined by
399 the frontend (like "add", "edit", "delete" or "preview"); it can be an action defined by a plugin
400 loaded by the frontend; or it can be a so called "server side action" (see "customActions" below)
402 customActions: ``Dict[str, dict]``
403 (Optional) A mapping of names of server-defined actions that can be used
404 in the ``actions`` list above to their definition dictionary. See .... for more details.
406 disabledActions: ``List[str, dict]``
407 (Optional) A list of disabled actions. The frontend will inject default actions like add or edit
408 even if they're not listed in actions. Listing them here will prevent that. It's up to the frontend
409 to decide if that action won't be visible at all or it's button just being disabled.
411 sortIndex: ``int``
412 (Optional) Defines the order in which the modules will appear in the main bar in
413 ascrending order.
415 indexedBones: ``List[str]``
416 (Optional) List of bones, for which an (composite?) index exists in this
417 view. This allows the fronted to signal the user that a given list can be sorted or filtered by this
418 bone. If no additional filters are enforced by the
419 :meth:`listFilter<viur.core.prototypes.list.listFilter>` and ``filter`` is not set, this should be
420 all bones which are marked as indexed.
422 changeInvalidates: ``List[str]``
423 (Optional) A list of module-names which depend on the entities handled
424 from this module. This allows the frontend to invalidate any caches in these depended modules if the
425 data in this module changes. Example: This module may be a list-module handling the file_rootNode
426 entities for the file module, so a edit/add/deletion action on this module should be reflected in the
427 rootNode-selector in the file-module itself. In this case, this property should be set to ``["file"]``.
429 moduleGroup: ``str``
430 (Optional) If set, should be a key of a moduleGroup defined in .... .
432 editViews: ``Dict[str, t.Any]``
433 (Optional) If set, will embed another list-widget in the edit forms for
434 a given entity. See .... for more details.
436 If this is a function, it must take no parameters and return the dictionary as shown above. This
437 can be used to customize the appearance of the Vi/Admin to individual users.
438 """
440 def __init__(self, moduleName: str, modulePath: str, *args, **kwargs):
441 self.render = None # will be set to the appropriate render instance at runtime
442 self._cached_description = None # caching used by describe()
443 self.moduleName = moduleName # Name of this module (usually it's class name, e.g. "file")
444 self.modulePath = modulePath # Path to this module in URL-routing (e.g. "json/file")
446 if self.handler and self.accessRights: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true
447 for right in self.accessRights:
448 right = f"{self.moduleName}-{right}"
450 # fixme: Turn conf.user.access_rights into a set.
451 if right not in conf.user.access_rights:
452 conf.user.access_rights.append(right)
454 # Collect methods and (sub)modules
455 self._methods = {}
456 self._modules = {}
457 self._update_methods()
459 def _update_methods(self):
460 """
461 Internal function to update methods and submodules.
462 This function should only be called when member attributes are dynamically modified by the module.
463 """
464 self._methods.clear()
465 self._modules.clear()
467 for key in dir(self):
468 if key[0] == "_":
469 continue
470 if isinstance(getattr(self.__class__, key, None), (property, functools.cached_property)): 470 ↛ 471line 470 didn't jump to line 471 because the condition on line 470 was never true
471 continue
473 prop = getattr(self, key)
475 if isinstance(prop, Method):
476 self._methods[key] = prop
477 elif isinstance(prop, Module): 477 ↛ 478line 477 didn't jump to line 478 because the condition on line 477 was never true
478 self._modules[key] = prop
480 def describe(self) -> dict | None:
481 """
482 Meta description of this module.
483 """
484 # Use cached description?
485 if isinstance(self._cached_description, dict):
486 return self._cached_description
488 # Retrieve handler
489 if not (handler := self.handler):
490 return None
492 # Default description
493 ret = {
494 "name": self.__class__.__name__,
495 "handler": ".".join((handler, self.__class__.__name__.lower())),
496 "methods": {
497 name: method.describe() for name, method in self._methods.items()
498 },
499 }
501 # Extend indexes, if available
502 # todo: This must be handled by SkelModule
503 if indexes := getattr(self, "indexes", None):
504 ret["indexes"] = indexes
506 # Merge adminInfo if present
507 if admin_info := self.adminInfo() if callable(self.adminInfo) else self.adminInfo:
508 assert isinstance(admin_info, dict), \
509 f"adminInfo can either be a dict or a callable returning a dict, but got {type(admin_info)}"
510 ret |= admin_info
512 # Cache description for later re-use.
513 if self._cached_description is not False:
514 self._cached_description = ret
516 return ret
518 def register(self, target: dict, render: object):
519 """
520 Registers this module's public functions to a given resolver.
521 This function is executed on start-up, and can be sub-classed.
522 """
523 # connect instance to render
524 self.render = render
526 # Map module under SEO-mapped name, if available.
527 if self.seo_language_map:
528 for lang in conf.i18n.available_languages or [conf.i18n.default_language]:
529 # Map the module under each translation
530 if translated_module_name := self.seo_language_map.get(lang):
531 translated_module = target.setdefault(translated_module_name, {})
533 # Map module methods to the previously determined target
534 for name, method in self._methods.items():
535 method.register(translated_module, name, lang)
537 conf.i18n.language_module_map[self.moduleName] = self.seo_language_map
539 # Map the module also under it's original name
540 if self.moduleName != "index":
541 target = target.setdefault(self.moduleName, {})
543 # Map module methods to the previously determined target
544 for name, method in self._methods.items():
545 method.register(target, name)
547 # Register sub modules
548 for name, module in self._modules.items():
549 module.register(target, self.render)