Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/module.py: 14%

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 

10 

11 

12class Method: 

13 """ 

14 Abstraction wrapper for any public available method. 

15 """ 

16 

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 

26 

27 return cls(func) 

28 

29 def __init__(self, func: t.Callable): 

30 # Content 

31 self._func = func 

32 self.__name__ = func.__name__ 

33 self._instance = None 

34 

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 

43 

44 # Inspection 

45 self.signature = inspect.signature(self._func) 

46 

47 # Guards 

48 self.guards = [] 

49 

50 def __get__(self, obj, objtype=None): 

51 """ 

52 This binds the Method to an object. 

53 

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 

60 

61 return self 

62 

63 def __call__(self, *args, **kwargs): 

64 """ 

65 Calls the method with given args and kwargs. 

66 

67 Prepares and filters argument values from args and kwargs regarding self._func's signature and type annotations, 

68 if present. 

69 

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. 

72 

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

79 

80 if trace := conf.debug.trace: 

81 logging.debug(f"calling {self._func=} with raw {args=}, {kwargs=}") 

82 

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

89 

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

101 return None 

102 raise ValueError(f"Expected None for parameter {name}. Got: {value!r}") 

103 

104 # complex types 

105 origin_type = t.get_origin(annotation) 

106 

107 if origin_type is list and len(annotation.__args__) == 1: 

108 if not isinstance(value, list): 

109 value = [value] 

110 

111 return [parse_value_by_annotation(annotation.__args__[0], name, item) for item in value] 

112 

113 elif origin_type is tuple and len(annotation.__args__) == 1: 

114 if not isinstance(value, tuple): 

115 value = (value, ) 

116 

117 return tuple(parse_value_by_annotation(annotation.__args__[0], name, item) for item in value) 

118 

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

122 

123 return value 

124 

125 elif origin_type is t.Union or isinstance(annotation, types.UnionType): 

126 for i, sub_annotation in enumerate(annotation.__args__): 

127 try: 

128 return parse_value_by_annotation(sub_annotation, name, value) 

129 except ValueError: 

130 if i == len(annotation.__args__) - 1: 

131 raise 

132 

133 elif annotation is db.Key: 

134 if isinstance(value, db.Key): 

135 return value 

136 elif isinstance(value, str): # Maybe we have an url encoded Key 

137 try: 

138 return db.Key.from_legacy_urlsafe(value) 

139 except Exception: 

140 pass 

141 return parse_value_by_annotation(int | str, name, value) 

142 

143 elif isinstance(annotation, enum.EnumMeta): 

144 try: 

145 return annotation(value) 

146 except ValueError as exc: 

147 for value_, member in annotation._value2member_map_.items(): 

148 if str(value) == str(value_): # Do a string comparison, it could be a IntEnum 

149 return member 

150 raise errors.NotAcceptable(f"{' '.join(exc.args)} for {name}") from exc 

151 

152 raise errors.NotAcceptable(f"Unhandled type {annotation=} for {name}={value!r}") 

153 

154 # examine parameters 

155 args_iter = iter(args) 

156 

157 parsed_args = [] 

158 parsed_kwargs = {} 

159 varargs = [] 

160 varkwargs = False 

161 

162 for i, (param_name, param) in enumerate(self.signature.parameters.items()): 

163 if self._instance and i == 0 and param_name == "self": 

164 continue 

165 

166 param_type = param.annotation 

167 param_required = param.default is param.empty 

168 

169 # take positional parameters first 

170 if param.kind in ( 

171 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

172 inspect.Parameter.POSITIONAL_ONLY 

173 ): 

174 try: 

175 value = next(args_iter) 

176 

177 if param_type is not param.empty: 

178 value = parse_value_by_annotation(param_type, param_name, value) 

179 

180 parsed_args.append(value) 

181 continue 

182 except StopIteration: 

183 pass 

184 

185 # otherwise take kwargs or variadics 

186 if ( 

187 param.kind in ( 

188 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

189 inspect.Parameter.KEYWORD_ONLY 

190 ) 

191 and param_name in kwargs 

192 ): 

193 value = kwargs.pop(param_name) 

194 

195 if param_type is not param.empty: 

196 try: 

197 value = parse_value_by_annotation(param_type, param_name, value) 

198 except ValueError as exc: 

199 raise errors.NotAcceptable(f"Invalid value for {param_name}") from exc 

200 

201 parsed_kwargs[param_name] = value 

202 

203 elif param.kind == inspect.Parameter.VAR_POSITIONAL: 

204 varargs = list(args_iter) 

205 elif param.kind == inspect.Parameter.VAR_KEYWORD: 

206 varkwargs = True 

207 elif param_required: 

208 if self.skey and param_name == self.skey["forward_payload"]: 

209 continue 

210 

211 raise errors.NotAcceptable(f"Missing required parameter {param_name!r}") 

212 

213 # Here's a short clarification on the variables used here: 

214 # 

215 # - parsed_args = tuple of (the type-parsed) arguments that have been assigned based on the signature 

216 # - parsed_kwargs = dict of (the type-parsed) keyword arguments that have been assigned based on the signature 

217 # - args = either parsed_args, or parsed_args + remaining args if the function accepts *args 

218 # - kwargs = either parsed_kwars, or parsed_kwargs | remaining kwargs if the function accepts **kwargs 

219 # - varargs = indicator that the args also contain variable args (*args) 

220 # - varkwargs = indicator that variable kwargs (**kwargs) are also contained in the kwargs 

221 # 

222 

223 # Extend args to any varargs, and redefine args 

224 args = tuple(parsed_args + varargs) 

225 

226 # always take "skey"-parameter name, when configured, as parsed_kwargs 

227 if self.skey and self.skey["name"] in kwargs: 

228 parsed_kwargs[self.skey["name"]] = kwargs.pop(self.skey["name"]) 

229 

230 # When varkwargs are accepted, merge parsed_kwargs and kwargs, otherwise just use parsed_kwargs 

231 if varkwargs := varkwargs and bool(kwargs): 

232 kwargs = parsed_kwargs | kwargs 

233 else: 

234 kwargs = parsed_kwargs 

235 

236 # Trace message for final call configuration 

237 if conf.debug.trace: 

238 logging.debug(f"calling {self._func=} with cleaned {args=}, {kwargs=}") 

239 # call decorators in reversed because they are added in the reversed order 

240 for func in reversed(self.guards): 

241 func(args=args, kwargs=kwargs, varargs=varargs, varkwargs=varkwargs) 

242 

243 # call with instance when provided 

244 if self._instance: 

245 return self._func(self._instance, *args, **kwargs) 

246 

247 return self._func(*args, **kwargs) 

248 

249 def describe(self) -> dict: 

250 """ 

251 Describes the Method with a 

252 """ 

253 return_doc = t.get_type_hints(self._func).get("return") 

254 

255 return { 

256 "args": { 

257 param.name: { 

258 "type": str(param.annotation) if param.annotation is not inspect.Parameter.empty else None, 

259 "default": str(param.default) if param.default is not inspect.Parameter.empty else None, 

260 } 

261 for param in self.signature.parameters.values() 

262 }, 

263 "returns": str(return_doc).strip() if return_doc else None, 

264 "accepts": self.methods, 

265 "docs": self._func.__doc__.strip() if self._func.__doc__ else None, 

266 "aliases": tuple(self.seo_language_map.keys()) if self.seo_language_map else None, 

267 } | self.additional_descr 

268 

269 def register(self, target: dict, name: str, language: str | None = None): 

270 """ 

271 Registers the Method under `name` and eventually some customized SEO-name for the provided language 

272 """ 

273 if self.exposed is None: 

274 return 

275 

276 target[name] = self 

277 

278 # reassign for SEO mapping as well 

279 if self.seo_language_map: 

280 for lang in tuple(self.seo_language_map.keys()) if not language else (language, ): 

281 if translated_name := self.seo_language_map.get(lang): 

282 target[translated_name] = self 

283 

284 

285class Module: 

286 """ 

287 This is the root module prototype that serves a minimal module in the ViUR system without any other bindings. 

288 """ 

289 

290 handler: str | t.Callable = None 

291 """ 

292 This is the module's handler, respectively its type. 

293 Use the @property-decorator in specific Modules to construct the handler's value dynamically. 

294 A module without a handler setting cannot be described, so cannot be handled by admin-tools. 

295 """ 

296 

297 accessRights: tuple[str] = None 

298 """ 

299 If set, a tuple of access rights (like add, edit, delete) that this module supports. 

300 

301 These will be prefixed on instance startup with the actual module name (becoming file-add, file-edit etc) 

302 and registered in ``conf.user.access_rights`` so these will be available on the access bone in user/add 

303 or user/edit. 

304 """ 

305 

306 roles: dict = {} 

307 r""" 

308 Allows to specify role settings for a module. 

309 

310 Defaults to no role definition, which ignores the module entirely in the role-system. 

311 In this case, access rights can still be set individually on the user's access bone. 

312 

313 A "*" wildcard can either be used as key or as value to allow for "all roles", or "all rights". 

314 

315 .. code-block:: python 

316 

317 # Example 

318 roles = { 

319 "*": "view", # Any role may only "view" 

320 "editor": ("add", "edit"), # Role "editor" may "add" or "edit", but not "delete" 

321 "admin": "*", # Role "admin" can do everything 

322 } 

323 

324 """ 

325 

326 seo_language_map: dict[str: str] = {} 

327 r""" 

328 The module name is the first part of a URL. 

329 SEO-identifiers have to be set as class-attribute ``seo_language_map`` of type ``dict[str, str]`` in the module. 

330 It maps a *language* to the according *identifier*. 

331 

332 .. code-block:: python 

333 :name: module seo-map 

334 :caption: modules/myorders.py 

335 :emphasize-lines: 4-7 

336 

337 from viur.core.prototypes import List 

338 

339 class MyOrders(List): 

340 seo_language_map = { 

341 "de": "bestellungen", 

342 "en": "orders", 

343 } 

344 

345 By default the module would be available under */myorders*, the lowercase module name. 

346 With the defined :attr:`seo_language_map`, it will become available as */de/bestellungen* and */en/orders*. 

347 

348 Great, this part is now user and robot friendly :) 

349 """ 

350 

351 adminInfo: dict[str, t.Any] | t.Callable = None 

352 """ 

353 This is a ``dict`` holding the information necessary for the Vi/Admin to handle this module. 

354 

355 name: ``str`` 

356 Human-readable module name that will be shown in the admin tool. 

357 

358 handler: ``str`` (``list``, ``tree`` or ``singleton``): 

359 Allows to override the handler provided by the module. Set this only when *really* necessary, 

360 otherwise it can be left out and is automatically injected by the Module's prototype. 

361 

362 icon: ``str`` 

363 (Optional) Either the Shoelace icon library name or a path relative to the project's deploy folder 

364 (e.g. /static/icons/viur.svg) for the icon used in the admin tool for this module. 

365 

366 columns: ``List[str]`` 

367 (Optional) List of columns (bone names) that are displayed by default. 

368 Used only by the List handler. 

369 

370 filter: ``Dict[str, str]`` 

371 (Optional) Dictionary of additional parameters that will be send along when 

372 fetching entities from the server. Can be used to filter the entities being displayed on the 

373 client-side. 

374 

375 display: ``str`` ("default", "hidden" or "group") 

376 (Optional) "hidden" will hide the module in the admin tool's main bar. 

377 (itwill not be accessible directly, however it's registered with the frontend so it can be used in a 

378 relational bone). "group" will show this module in the main bar, but it will not be clickable. 

379 Clicking it will just try to expand it (assuming there are additional views defined). 

380 

381 preview: ``Union[str, Dict[str, str]]`` 

382 (Optional) A url that will be opened in a new tab and is expected to display 

383 the entity selected in the table. Can be “/{{module}}/view/{{key}}", with {{module}} and {{key}} getting 

384 replaced as needed. If more than one preview-url is needed, supply a dictionary where the key is 

385 the URL and the value the description shown to the user. 

386 

387 views: ``List[Dict[str, t.Any]]`` 

388 (Optional) List of nested adminInfo like dictionaries. Used to define 

389 additional views on the module. Useful f.e. for an order module, where you want separate list of 

390 "payed orders", "unpayed orders", "orders waiting for shipment", etc. If such views are defined, 

391 the top-level entry in the menu bar will expand if clicked, revealing these additional filters. 

392 

393 actions: ``List[str]`` 

394 (Optional) List of actions supported by this modules. Actions can be defined by 

395 the frontend (like "add", "edit", "delete" or "preview"); it can be an action defined by a plugin 

396 loaded by the frontend; or it can be a so called "server side action" (see "customActions" below) 

397 

398 customActions: ``Dict[str, dict]`` 

399 (Optional) A mapping of names of server-defined actions that can be used 

400 in the ``actions`` list above to their definition dictionary. See .... for more details. 

401 

402 disabledActions: ``List[str, dict]`` 

403 (Optional) A list of disabled actions. The frontend will inject default actions like add or edit 

404 even if they're not listed in actions. Listing them here will prevent that. It's up to the frontend 

405 to decide if that action won't be visible at all or it's button just being disabled. 

406 

407 sortIndex: ``int`` 

408 (Optional) Defines the order in which the modules will appear in the main bar in 

409 ascrending order. 

410 

411 indexedBones: ``List[str]`` 

412 (Optional) List of bones, for which an (composite?) index exists in this 

413 view. This allows the fronted to signal the user that a given list can be sorted or filtered by this 

414 bone. If no additional filters are enforced by the 

415 :meth:`listFilter<viur.core.prototypes.list.listFilter>` and ``filter`` is not set, this should be 

416 all bones which are marked as indexed. 

417 

418 changeInvalidates: ``List[str]`` 

419 (Optional) A list of module-names which depend on the entities handled 

420 from this module. This allows the frontend to invalidate any caches in these depended modules if the 

421 data in this module changes. Example: This module may be a list-module handling the file_rootNode 

422 entities for the file module, so a edit/add/deletion action on this module should be reflected in the 

423 rootNode-selector in the file-module itself. In this case, this property should be set to ``["file"]``. 

424 

425 moduleGroup: ``str`` 

426 (Optional) If set, should be a key of a moduleGroup defined in .... . 

427 

428 editViews: ``Dict[str, t.Any]`` 

429 (Optional) If set, will embed another list-widget in the edit forms for 

430 a given entity. See .... for more details. 

431 

432 If this is a function, it must take no parameters and return the dictionary as shown above. This 

433 can be used to customize the appearance of the Vi/Admin to individual users. 

434 """ 

435 

436 def __init__(self, moduleName: str, modulePath: str, *args, **kwargs): 

437 self.render = None # will be set to the appropriate render instance at runtime 

438 self._cached_description = None # caching used by describe() 

439 self.moduleName = moduleName # Name of this module (usually it's class name, e.g. "file") 

440 self.modulePath = modulePath # Path to this module in URL-routing (e.g. "json/file") 

441 

442 if self.handler and self.accessRights: 

443 for right in self.accessRights: 

444 right = f"{self.moduleName}-{right}" 

445 

446 # fixme: Turn conf.user.access_rights into a set. 

447 if right not in conf.user.access_rights: 

448 conf.user.access_rights.append(right) 

449 

450 # Collect methods and (sub)modules 

451 self._methods = {} 

452 self._modules = {} 

453 self._update_methods() 

454 

455 def _update_methods(self): 

456 """ 

457 Internal function to update methods and submodules. 

458 This function should only be called when member attributes are dynamically modified by the module. 

459 """ 

460 self._methods.clear() 

461 self._modules.clear() 

462 

463 for key in dir(self): 

464 if key[0] == "_": 

465 continue 

466 if isinstance(getattr(self.__class__, key, None), (property, functools.cached_property)): 

467 continue 

468 

469 prop = getattr(self, key) 

470 

471 if isinstance(prop, Method): 

472 self._methods[key] = prop 

473 elif isinstance(prop, Module): 

474 self._modules[key] = prop 

475 

476 def describe(self) -> dict | None: 

477 """ 

478 Meta description of this module. 

479 """ 

480 # Use cached description? 

481 if isinstance(self._cached_description, dict): 

482 return self._cached_description 

483 

484 # Retrieve handler 

485 if not (handler := self.handler): 

486 return None 

487 

488 # Default description 

489 ret = { 

490 "name": self.__class__.__name__, 

491 "handler": ".".join((handler, self.__class__.__name__.lower())), 

492 "methods": { 

493 name: method.describe() for name, method in self._methods.items() 

494 }, 

495 } 

496 

497 # Extend indexes, if available 

498 # todo: This must be handled by SkelModule 

499 if indexes := getattr(self, "indexes", None): 

500 ret["indexes"] = indexes 

501 

502 # Merge adminInfo if present 

503 if admin_info := self.adminInfo() if callable(self.adminInfo) else self.adminInfo: 

504 assert isinstance(admin_info, dict), \ 

505 f"adminInfo can either be a dict or a callable returning a dict, but got {type(admin_info)}" 

506 ret |= admin_info 

507 

508 # Cache description for later re-use. 

509 if self._cached_description is not False: 

510 self._cached_description = ret 

511 

512 return ret 

513 

514 def register(self, target: dict, render: object): 

515 """ 

516 Registers this module's public functions to a given resolver. 

517 This function is executed on start-up, and can be sub-classed. 

518 """ 

519 # connect instance to render 

520 self.render = render 

521 

522 # Map module under SEO-mapped name, if available. 

523 if self.seo_language_map: 

524 for lang in conf.i18n.available_languages or [conf.i18n.default_language]: 

525 # Map the module under each translation 

526 if translated_module_name := self.seo_language_map.get(lang): 

527 translated_module = target.setdefault(translated_module_name, {}) 

528 

529 # Map module methods to the previously determined target 

530 for name, method in self._methods.items(): 

531 method.register(translated_module, name, lang) 

532 

533 conf.i18n.language_module_map[self.moduleName] = self.seo_language_map 

534 

535 # Map the module also under it's original name 

536 if self.moduleName != "index": 

537 target = target.setdefault(self.moduleName, {}) 

538 

539 # Map module methods to the previously determined target 

540 for name, method in self._methods.items(): 

541 method.register(target, name) 

542 

543 # Register sub modules 

544 for name, module in self._modules.items(): 

545 module.register(target, self.render)