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

166 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-09-25 15:02 +0000

1""" 

2ViUR-core 

3Copyright © 2026 Mausbrand Informationssysteme GmbH 

4 

5https://core.docs.viur.dev 

6Licensed under the MIT license. See LICENSE for more information. 

7""" 

8import fnmatch 

9import os 

10import sys 

11 

12# Set a dummy project id to survive API Client initializations 

13if sys.argv[0].endswith("viur-migrate"): # FIXME: What a "kinda hackish" solution... 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true

14 os.environ["GOOGLE_CLOUD_PROJECT"] = "dummy" 

15 

16from google.appengine.api import wrap_wsgi_app 

17from types import ModuleType 

18from viur.core import i18n, request, utils 

19from viur.core.config import conf 

20from viur.core.decorators import access, exposed, force_post, force_ssl, internal_exposed, skey 

21from viur.core.request import before_request, after_request 

22from viur.core.i18n import translate 

23from viur.core.module import Method, Module 

24import inspect 

25import typing as t 

26import warnings 

27from .tasks import ( 

28 callDeferred, 

29 CallDeferred, 

30 DeleteEntitiesIter, 

31 PeriodicTask, 

32 QueryIter, 

33 retry_n_times, 

34 runStartupTasks, 

35 StartupTask, 

36 TaskHandler, 

37) 

38 

39if not sys.argv[0].endswith("viur-migrate"): # FIXME: What a "kinda hackish" solution... 39 ↛ 43line 39 didn't jump to line 43 because the condition on line 39 was always true

40 # noinspection PyUnresolvedReferences 

41 from viur.core import logging as viurLogging # unused import, must exist, initializes request logging 

42 

43import logging # this import has to stay here, see #571 

44from deprecated.sphinx import deprecated 

45 

46__all__ = [ 

47 # basics from this __init__ 

48 "setDefaultLanguage", 

49 "setDefaultDomainLanguage", 

50 "setup", 

51 # prototypes 

52 "Module", 

53 "Method", 

54 # tasks 

55 "DeleteEntitiesIter", 

56 "QueryIter", 

57 "retry_n_times", 

58 "callDeferred", 

59 "CallDeferred", 

60 "StartupTask", 

61 "PeriodicTask", 

62 # Decorators 

63 "access", 

64 "after_request", 

65 "before_request", 

66 "exposed", 

67 "force_post", 

68 "force_ssl", 

69 "internal_exposed", 

70 "skey", 

71 # others 

72 "conf", 

73 "translate", 

74] 

75 

76# Show DeprecationWarning from the viur-core 

77warnings.filterwarnings("once", category=DeprecationWarning) 

78warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"viur\.datastore.*", 

79 message="'clonedBoneMap' was renamed into 'bone_map'") 

80 

81 

82@deprecated( 

83 version="3.8.0", 

84 reason="Simply set `conf.i18n.default_language` to the desired language." 

85) 

86def setDefaultLanguage(lang: str): 

87 """ 

88 Sets the default language used by ViUR to *lang*. 

89 

90 :param lang: Name of the language module to use by default. 

91 """ 

92 msg = f"`setDefaultLanguage(\"{lang}\")` is deprecated; " \ 

93 f"Replace the call by `conf.i18n.default_language = \"{lang.lower()}\"`" 

94 warnings.warn(msg, DeprecationWarning, stacklevel=2) 

95 logging.warning(msg) 

96 

97 conf.i18n.default_language = lang.lower() 

98 

99 

100def setDefaultDomainLanguage(domain: str, lang: str): 

101 """ 

102 If conf.i18n.language_method is set to "domain", this function allows setting the map of which domain 

103 should use which language. 

104 :param domain: The domain for which the language should be set 

105 :param lang: The language to use (in ISO2 format, e.g. "DE") 

106 """ 

107 host = domain.lower().strip(" /") 

108 if host.startswith("www."): 

109 host = host[4:] 

110 conf.i18n.domain_language_mapping[host] = lang.lower() 

111 

112 

113def __build_app(modules: ModuleType | object, renderers: ModuleType | object, default: str = None) -> Module: 

114 """ 

115 Creates the application-context for the current instance. 

116 

117 This function converts the classes found in the *modules*-module, 

118 and the given renders into the object found at ``conf.main_app``. 

119 

120 Every class found in *modules* becomes 

121 

122 - instanced 

123 - get the corresponding renderer attached 

124 - will be attached to ``conf.main_app`` 

125 

126 :param modules: Usually the module provided as *modules* directory within the application. 

127 :param renderers: Usually the module *viur.core.renders*, or a dictionary renderName => renderClass. 

128 :param default: Name of the renderer, which will form the root of the application. 

129 This will be the renderer, which wont get a prefix, usually html. 

130 (=> /user instead of /html/user) 

131 """ 

132 if not isinstance(renderers, dict): 

133 # build up the dict from viur.core.render 

134 renderers, mod = {}, renderers 

135 

136 from viur.core.render.abstract import AbstractRenderer 

137 

138 for render_name, render_mod in vars(mod).items(): 

139 if inspect.ismodule(render_mod): 

140 for render_clsname, render_cls in vars(render_mod).items(): 

141 # this is "kinda hackish..." because ViUR 3's current renderer concept is pure bulls*t... 

142 if render_clsname == "DefaultRender": 

143 continue 

144 

145 if ( 

146 # test for a renderer 

147 (inspect.isclass(render_cls) and issubclass(render_cls, AbstractRenderer)) 

148 # bullsh*t, this must be entirely reworked! 

149 or render_clsname == "_postProcessAppObj" 

150 ): 

151 renderers.setdefault(render_name, {}) 

152 renderers[render_name][render_clsname] = render_cls 

153 

154 # assign ViUR system modules 

155 from viur.core.modules.moduleconf import ModuleConf # noqa: E402 # import works only here because circular imports 

156 from viur.core.modules.script import Script # noqa: E402 # import works only here because circular imports 

157 from viur.core.modules.translation import Translation # noqa: E402 # import works only here because circular imports 

158 from viur.core.prototypes.instanced_module import InstancedModule # noqa: E402 # import works only here because circular imports 

159 

160 for name, cls in { 

161 "_tasks": TaskHandler, 

162 "_moduleconf": ModuleConf, 

163 "_translation": Translation, 

164 "script": Script, 

165 }.items(): 

166 # Check whether name is contained in modules so that it can be overwritten 

167 if name not in vars(modules): 

168 setattr(modules, name, cls) 

169 

170 assert issubclass(getattr(modules, name), cls) 

171 

172 # Resolver defines the URL mapping 

173 resolver = {} 

174 

175 # Index is mapping all module instances for global access 

176 index = (modules.index if hasattr(modules, "index") else Module)("index", "") 

177 index.register(resolver, renderers[default]["default"](parent=index)) 

178 

179 for module_name, module_cls in vars(modules).items(): # iterate over all modules 

180 if module_name == "index": 

181 continue # ignore index, as it has been processed before! 

182 

183 if module_name in renderers: 

184 raise NameError(f"Cannot name module {module_name!r}, as it is a reserved render's name") 

185 

186 if not ( # we define the cases we want to use and then negate them all 

187 (inspect.isclass(module_cls) and issubclass(module_cls, Module) # is a normal Module class 

188 and not issubclass(module_cls, InstancedModule)) # but not a "instantiable" Module 

189 or isinstance(module_cls, InstancedModule) # is an already instanced Module 

190 ): 

191 continue 

192 

193 # remember module_instance for default renderer. 

194 module_instance = default_module_instance = None 

195 

196 for render_name, render in renderers.items(): # look, if a particular renderer should be built 

197 # Only continue when module_cls is configured for this render 

198 # todo: VIUR4 this is for legacy reasons, can be done better! 

199 if not getattr(module_cls, render_name, False): 

200 continue 

201 

202 # Create a new module instance 

203 module_instance = module_cls( 

204 module_name, ("/" + render_name if render_name != default else "") + "/" + module_name 

205 ) 

206 

207 # Attach the module-specific or the default render 

208 if render_name == default: # default or render (sub)namespace? 

209 default_module_instance = module_instance 

210 target = resolver 

211 else: 

212 if getattr(index, render_name, True) is True: 

213 # Render is not build yet, or it is just the simple marker that a given render should be build 

214 setattr(index, render_name, Module(render_name, "/" + render_name)) 

215 

216 # Attach the module to the given renderer node 

217 setattr(getattr(index, render_name), module_name, module_instance) 

218 target = resolver.setdefault(render_name, {}) 

219 

220 module_instance.register(target, render.get(module_name, render["default"])(parent=module_instance)) 

221 

222 # Apply Renderers postProcess Filters 

223 if "_postProcessAppObj" in render: # todo: This is ugly! 

224 render["_postProcessAppObj"](target) 

225 

226 # Ugly solution, but there is no better way to do it in ViUR 3: 

227 # Allow that any module can be accessed by `conf.main_app.<modulename>`, 

228 # either with default render or the last created render. 

229 # This behavior does NOT influence the routing. 

230 if default_module_instance or module_instance: 

231 setattr(index, module_name, default_module_instance or module_instance) 

232 

233 # fixme: Below is also ugly... 

234 if default in renderers and hasattr(renderers[default]["default"], "renderEmail"): 

235 conf.emailRenderer = renderers[default]["default"]().renderEmail 

236 elif "html" in renderers: 

237 conf.emailRenderer = renderers["html"]["default"]().renderEmail 

238 

239 # This might be useful for debugging, please keep it for now. 

240 if conf.debug.trace: 

241 import pprint 

242 logging.debug(pprint.pformat(resolver)) 

243 

244 conf.main_resolver = resolver 

245 conf.main_app = index 

246 

247 

248def setup(modules: ModuleType | object, render: ModuleType | object = None, default: str = "html"): 

249 """ 

250 Define whats going to be served by this instance. 

251 

252 :param modules: Usually the module provided as *modules* directory within the application. 

253 :param render: Usually the module *viur.core.renders*, or a dictionary renderName => renderClass. 

254 :param default: Name of the renderer, which will form the root of the application.\ 

255 This will be the renderer, which wont get a prefix, usually html. \ 

256 (=> /user instead of /html/user) 

257 """ 

258 from viur.core.bones.base import setSystemInitialized 

259 # noinspection PyUnresolvedReferences 

260 import skeletons # This import is not used here but _must_ remain to ensure that the 

261 # application's data models are explicitly imported at some place! 

262 for application_id in conf.valid_application_ids: 

263 if fnmatch.fnmatch(conf.instance.project_id, application_id): 

264 break 

265 else: 

266 raise RuntimeError( 

267 f"""Refusing to start, {conf.instance.project_id=} is not in {conf.valid_application_ids=}""") 

268 if not render: 

269 import viur.core.render 

270 render = viur.core.render 

271 

272 __build_app(modules, render, default) 

273 

274 # Send warning email in case trace is activated in a cloud environment 

275 if ((conf.debug.trace 

276 or conf.debug.trace_external_call_routing 

277 or conf.debug.trace_internal_call_routing) 

278 and (not conf.instance.is_dev_server or conf.debug.dev_server_cloud_logging)): 

279 from viur.core import email 

280 try: 

281 email.send_email_to_admins( 

282 "Debug mode enabled", 

283 "ViUR just started a new Instance with call tracing enabled! This might log sensitive information!" 

284 ) 

285 except Exception as exc: # OverQuota, whatever 

286 logging.exception(exc) 

287 # Ensure that our Content Security Policy Header Cache gets build 

288 from viur.core import securityheaders 

289 securityheaders._rebuildCspHeaderCache() 

290 securityheaders._rebuildPermissionHeaderCache() 

291 setSystemInitialized() 

292 # Assert that all security related headers are in a sane state 

293 if conf.security.content_security_policy and conf.security.content_security_policy["_headerCache"]: 

294 for k in conf.security.content_security_policy["_headerCache"]: 

295 if not k.startswith("Content-Security-Policy"): 

296 raise AssertionError("Got unexpected header in " 

297 "conf.security.content_security_policy['_headerCache']") 

298 if conf.security.strict_transport_security: 

299 if not conf.security.strict_transport_security.startswith("max-age"): 

300 raise AssertionError("Got unexpected header in conf.security.strict_transport_security") 

301 securityheaders._validate_reporting_config() 

302 crossDomainPolicies = {None, "none", "master-only", "by-content-type", "all"} 

303 if conf.security.x_permitted_cross_domain_policies not in crossDomainPolicies: 

304 raise AssertionError("conf.security.x_permitted_cross_domain_policies " 

305 f"must be one of {crossDomainPolicies!r}") 

306 if conf.security.x_frame_options is not None and isinstance(conf.security.x_frame_options, tuple): 

307 mode, uri = conf.security.x_frame_options 

308 assert mode in ["deny", "sameorigin", "allow-from"] 

309 if mode == "allow-from": 

310 assert uri is not None and (uri.lower().startswith("https://") or uri.lower().startswith("http://")) 

311 runStartupTasks() # Add a deferred call to run all queued startup tasks 

312 i18n.initializeTranslations() 

313 if conf.file_hmac_key is None: 

314 from viur.core import db 

315 key = db.Key("viur-conf", "viur-conf") 

316 if not (obj := db.get(key)): # create a new "viur-conf"? 

317 logging.info("Creating new viur-conf") 

318 obj = db.Entity(key) 

319 

320 if "hmacKey" not in obj: # create a new hmacKey 

321 logging.info("Creating new hmacKey") 

322 obj["hmacKey"] = utils.string.random(length=20) 

323 db.put(obj) 

324 

325 conf.file_hmac_key = bytes(obj["hmacKey"], "utf-8") 

326 

327 if conf.instance.is_dev_server: 

328 WIDTH = 80 # defines the standard width 

329 FILL = "#" # define sthe fill char (must be len(1)!) 

330 PYTHON_VERSION = (sys.version_info.major, sys.version_info.minor, sys.version_info.micro) 

331 

332 # extra banner lines for a non-default datastore target 

333 datastore_lines = [] 

334 if conf.db.name: 

335 datastore_lines.append(f"""database = \033[1;33m{conf.db.name}\033[0m""") 

336 if conf.db.namespace: 

337 datastore_lines.append(f"""namespace = \033[1;33m{conf.db.namespace}\033[0m""") 

338 

339 # define lines to show 

340 lines = ( 

341 " LOCAL DEVELOPMENT SERVER IS UP AND RUNNING ", # title line 

342 f"""project = \033[1;31m{conf.instance.project_id}\033[0m""", 

343 f"""python = \033[1;32m{".".join((str(i) for i in PYTHON_VERSION))}\033[0m""", 

344 f"""viur = \033[1;32m{".".join((str(i) for i in conf.version))}\033[0m""", 

345 *datastore_lines, # only when a non-default db/namespace is set 

346 "" # empty line 

347 ) 

348 

349 # first and last line are shown with a cool line made of FILL 

350 first_last = (0, len(lines) - 1) 

351 

352 # dump to console 

353 for i, line in enumerate(lines): 

354 print( 

355 f"""\033[0m{FILL}{line:{ 

356 FILL if i in first_last else " "}^{(WIDTH - 2) + (11 if i not in first_last else 0) 

357 }}{FILL}""" 

358 ) 

359 

360 return wrap_wsgi_app(app) 

361 

362 

363def app(environ: dict, start_response: t.Callable): 

364 return request.Router(environ).response(environ, start_response) 

365 

366 

367# DEPRECATED ATTRIBUTES HANDLING 

368 

369__DEPRECATED_DECORATORS = { 

370 # stuff prior viur-core < 3.5 

371 "forcePost": ("force_post", force_post), 

372 "forceSSL": ("force_ssl", force_ssl), 

373 "internalExposed": ("internal_exposed", internal_exposed) 

374} 

375 

376 

377def __getattr__(attr: str) -> object: 

378 if entry := __DEPRECATED_DECORATORS.get(attr): 378 ↛ 379line 378 didn't jump to line 379 because the condition on line 378 was never true

379 func = entry[1] 

380 msg = f"@{attr} was replaced by @{entry[0]}" 

381 warnings.warn(msg, DeprecationWarning, stacklevel=2) 

382 logging.warning(msg, stacklevel=2) 

383 return func 

384 

385 return super(__import__(__name__).__class__).__getattr__(attr)