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 

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

138 try: 

139 return db.normalize_key(value) 

140 except Exception: 

141 pass 

142 

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

144 

145 elif isinstance(annotation, enum.EnumMeta): 

146 try: 

147 return annotation(value) 

148 except ValueError as exc: 

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

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

151 return member 

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

153 

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

155 

156 # examine parameters 

157 args_iter = iter(args) 

158 

159 parsed_args = [] 

160 parsed_kwargs = {} 

161 varargs = [] 

162 varkwargs = False 

163 

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

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

166 continue 

167 

168 param_type = param.annotation 

169 param_required = param.default is param.empty 

170 

171 # take positional parameters first 

172 if param.kind in ( 

173 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

174 inspect.Parameter.POSITIONAL_ONLY 

175 ): 

176 try: 

177 value = next(args_iter) 

178 

179 if param_type is not param.empty: 

180 value = parse_value_by_annotation(param_type, param_name, value) 

181 

182 parsed_args.append(value) 

183 continue 

184 except StopIteration: 

185 pass 

186 

187 # otherwise take kwargs or variadics 

188 if ( 

189 param.kind in ( 

190 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

191 inspect.Parameter.KEYWORD_ONLY 

192 ) 

193 and param_name in kwargs 

194 ): 

195 value = kwargs.pop(param_name) 

196 

197 if param_type is not param.empty: 

198 try: 

199 value = parse_value_by_annotation(param_type, param_name, value) 

200 except ValueError as exc: 

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

202 

203 parsed_kwargs[param_name] = value 

204 

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

206 varargs = list(args_iter) 

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

208 varkwargs = True 

209 elif param_required: 

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

211 continue 

212 

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

214 

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

216 # 

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

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

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

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

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

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

223 # 

224 

225 # Extend args to any varargs, and redefine args 

226 args = tuple(parsed_args + varargs) 

227 

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

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

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

231 

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

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

234 kwargs = parsed_kwargs | kwargs 

235 else: 

236 kwargs = parsed_kwargs 

237 

238 # Trace message for final call configuration 

239 if conf.debug.trace: 

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

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

242 for func in reversed(self.guards): 

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

244 

245 # call with instance when provided 

246 if self._instance: 

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

248 

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

250 

251 def describe(self) -> dict: 

252 """ 

253 Describes the Method with a 

254 """ 

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

256 

257 return { 

258 "args": { 

259 param.name: { 

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

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

262 } 

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

264 }, 

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

266 "accepts": self.methods, 

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

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

269 } | self.additional_descr 

270 

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

272 """ 

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

274 """ 

275 if self.exposed is None: 

276 return 

277 

278 target[name] = self 

279 

280 # reassign for SEO mapping as well 

281 if self.seo_language_map: 

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

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

284 target[translated_name] = self 

285 

286 

287class Module: 

288 """ 

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

290 """ 

291 

292 handler: str | t.Callable = None 

293 """ 

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

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

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

297 """ 

298 

299 accessRights: tuple[str] = None 

300 """ 

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

302 

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

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

305 or user/edit. 

306 """ 

307 

308 roles: dict = {} 

309 r""" 

310 Allows to specify role settings for a module. 

311 

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

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

314 

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

316 

317 .. code-block:: python 

318 

319 # Example 

320 roles = { 

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

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

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

324 } 

325 

326 """ 

327 

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

329 r""" 

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

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

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

333 

334 .. code-block:: python 

335 :name: module seo-map 

336 :caption: modules/myorders.py 

337 :emphasize-lines: 4-7 

338 

339 from viur.core.prototypes import List 

340 

341 class MyOrders(List): 

342 seo_language_map = { 

343 "de": "bestellungen", 

344 "en": "orders", 

345 } 

346 

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

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

349 

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

351 """ 

352 

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

354 """ 

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

356 

357 name: ``str`` 

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

359 

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

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

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

363 

364 icon: ``str`` 

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

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

367 

368 columns: ``List[str]`` 

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

370 Used only by the List handler. 

371 

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

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

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

375 client-side. 

376 

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

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

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

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

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

382 

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

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

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

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

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

388 

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

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

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

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

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

394 

395 actions: ``List[str]`` 

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

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

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

399 

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

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

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

403 

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

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

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

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

408 

409 sortIndex: ``int`` 

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

411 ascrending order. 

412 

413 indexedBones: ``List[str]`` 

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

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

416 bone. If no additional filters are enforced by the 

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

418 all bones which are marked as indexed. 

419 

420 changeInvalidates: ``List[str]`` 

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

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

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

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

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

426 

427 moduleGroup: ``str`` 

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

429 

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

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

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

433 

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

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

436 """ 

437 

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

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

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

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

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

443 

444 if self.handler and self.accessRights: 

445 for right in self.accessRights: 

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

447 

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

449 if right not in conf.user.access_rights: 

450 conf.user.access_rights.append(right) 

451 

452 # Collect methods and (sub)modules 

453 self._methods = {} 

454 self._modules = {} 

455 self._update_methods() 

456 

457 def _update_methods(self): 

458 """ 

459 Internal function to update methods and submodules. 

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

461 """ 

462 self._methods.clear() 

463 self._modules.clear() 

464 

465 for key in dir(self): 

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

467 continue 

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

469 continue 

470 

471 prop = getattr(self, key) 

472 

473 if isinstance(prop, Method): 

474 self._methods[key] = prop 

475 elif isinstance(prop, Module): 

476 self._modules[key] = prop 

477 

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

479 """ 

480 Meta description of this module. 

481 """ 

482 # Use cached description? 

483 if isinstance(self._cached_description, dict): 

484 return self._cached_description 

485 

486 # Retrieve handler 

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

488 return None 

489 

490 # Default description 

491 ret = { 

492 "name": self.__class__.__name__, 

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

494 "methods": { 

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

496 }, 

497 } 

498 

499 # Extend indexes, if available 

500 # todo: This must be handled by SkelModule 

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

502 ret["indexes"] = indexes 

503 

504 # Merge adminInfo if present 

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

506 assert isinstance(admin_info, dict), \ 

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

508 ret |= admin_info 

509 

510 # Cache description for later re-use. 

511 if self._cached_description is not False: 

512 self._cached_description = ret 

513 

514 return ret 

515 

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

517 """ 

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

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

520 """ 

521 # connect instance to render 

522 self.render = render 

523 

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

525 if self.seo_language_map: 

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

527 # Map the module under each translation 

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

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

530 

531 # Map module methods to the previously determined target 

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

533 method.register(translated_module, name, lang) 

534 

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

536 

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

538 if self.moduleName != "index": 

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

540 

541 # Map module methods to the previously determined target 

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

543 method.register(target, name) 

544 

545 # Register sub modules 

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

547 module.register(target, self.render)