Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/__init__.py: 14%
168 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-12 02:43 +0000
1"""
2ViUR-core
3Copyright © 2026 Mausbrand Informationssysteme GmbH
5https://core.docs.viur.dev
6Licensed under the MIT license. See LICENSE for more information.
7"""
8import fnmatch
9import os
10import sys
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"
16from google.appengine.api import wrap_wsgi_app
17from google.appengine.runtime import initialize as gae_initialize
18from types import ModuleType
19from viur.core import i18n, request, utils
20from viur.core.config import conf
21from viur.core.decorators import access, exposed, force_post, force_ssl, internal_exposed, skey
22from viur.core.request import before_request, after_request
23from viur.core.i18n import translate
24from viur.core.module import Method, Module
25import inspect
26import typing as t
27import warnings
28from .tasks import (
29 callDeferred,
30 CallDeferred,
31 DeleteEntitiesIter,
32 PeriodicTask,
33 QueryIter,
34 retry_n_times,
35 runStartupTasks,
36 StartupTask,
37 TaskHandler,
38)
40if not sys.argv[0].endswith("viur-migrate"): # FIXME: What a "kinda hackish" solution... 40 ↛ 44line 40 didn't jump to line 44 because the condition on line 40 was always true
41 # noinspection PyUnresolvedReferences
42 from viur.core import logging as viurLogging # unused import, must exist, initializes request logging
44import logging # this import has to stay here, see #571
45from deprecated.sphinx import deprecated
47__all__ = [
48 # basics from this __init__
49 "setDefaultLanguage",
50 "setDefaultDomainLanguage",
51 "setup",
52 # prototypes
53 "Module",
54 "Method",
55 # tasks
56 "DeleteEntitiesIter",
57 "QueryIter",
58 "retry_n_times",
59 "callDeferred",
60 "CallDeferred",
61 "StartupTask",
62 "PeriodicTask",
63 # Decorators
64 "access",
65 "after_request",
66 "before_request",
67 "exposed",
68 "force_post",
69 "force_ssl",
70 "internal_exposed",
71 "skey",
72 # others
73 "conf",
74 "translate",
75]
77# Show DeprecationWarning from the viur-core
78warnings.filterwarnings("once", category=DeprecationWarning)
79warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"viur\.datastore.*",
80 message="'clonedBoneMap' was renamed into 'bone_map'")
83@deprecated(
84 version="3.8.0",
85 reason="Simply set `conf.i18n.default_language` to the desired language."
86)
87def setDefaultLanguage(lang: str):
88 """
89 Sets the default language used by ViUR to *lang*.
91 :param lang: Name of the language module to use by default.
92 """
93 msg = f"`setDefaultLanguage(\"{lang}\")` is deprecated; " \
94 f"Replace the call by `conf.i18n.default_language = \"{lang.lower()}\"`"
95 warnings.warn(msg, DeprecationWarning, stacklevel=2)
96 logging.warning(msg)
98 conf.i18n.default_language = lang.lower()
101def setDefaultDomainLanguage(domain: str, lang: str):
102 """
103 If conf.i18n.language_method is set to "domain", this function allows setting the map of which domain
104 should use which language.
105 :param domain: The domain for which the language should be set
106 :param lang: The language to use (in ISO2 format, e.g. "DE")
107 """
108 host = domain.lower().strip(" /")
109 if host.startswith("www."):
110 host = host[4:]
111 conf.i18n.domain_language_mapping[host] = lang.lower()
114def __build_app(modules: ModuleType | object, renderers: ModuleType | object, default: str = None) -> Module:
115 """
116 Creates the application-context for the current instance.
118 This function converts the classes found in the *modules*-module,
119 and the given renders into the object found at ``conf.main_app``.
121 Every class found in *modules* becomes
123 - instanced
124 - get the corresponding renderer attached
125 - will be attached to ``conf.main_app``
127 :param modules: Usually the module provided as *modules* directory within the application.
128 :param renderers: Usually the module *viur.core.renders*, or a dictionary renderName => renderClass.
129 :param default: Name of the renderer, which will form the root of the application.
130 This will be the renderer, which wont get a prefix, usually html.
131 (=> /user instead of /html/user)
132 """
133 if not isinstance(renderers, dict):
134 # build up the dict from viur.core.render
135 renderers, mod = {}, renderers
137 from viur.core.render.abstract import AbstractRenderer
139 for render_name, render_mod in vars(mod).items():
140 if inspect.ismodule(render_mod):
141 for render_clsname, render_cls in vars(render_mod).items():
142 # this is "kinda hackish..." because ViUR 3's current renderer concept is pure bulls*t...
143 if render_clsname == "DefaultRender":
144 continue
146 if (
147 # test for a renderer
148 (inspect.isclass(render_cls) and issubclass(render_cls, AbstractRenderer))
149 # bullsh*t, this must be entirely reworked!
150 or render_clsname == "_postProcessAppObj"
151 ):
152 renderers.setdefault(render_name, {})
153 renderers[render_name][render_clsname] = render_cls
155 # assign ViUR system modules
156 from viur.core.modules.moduleconf import ModuleConf # noqa: E402 # import works only here because circular imports
157 from viur.core.modules.script import Script # noqa: E402 # import works only here because circular imports
158 from viur.core.modules.translation import Translation # noqa: E402 # import works only here because circular imports
159 from viur.core.prototypes.instanced_module import InstancedModule # noqa: E402 # import works only here because circular imports
161 for name, cls in {
162 "_tasks": TaskHandler,
163 "_moduleconf": ModuleConf,
164 "_translation": Translation,
165 "script": Script,
166 }.items():
167 # Check whether name is contained in modules so that it can be overwritten
168 if name not in vars(modules):
169 setattr(modules, name, cls)
171 assert issubclass(getattr(modules, name), cls)
173 # Resolver defines the URL mapping
174 resolver = {}
176 # Index is mapping all module instances for global access
177 index = (modules.index if hasattr(modules, "index") else Module)("index", "")
178 index.register(resolver, renderers[default]["default"](parent=index))
180 for module_name, module_cls in vars(modules).items(): # iterate over all modules
181 if module_name == "index":
182 continue # ignore index, as it has been processed before!
184 if module_name in renderers:
185 raise NameError(f"Cannot name module {module_name!r}, as it is a reserved render's name")
187 if not ( # we define the cases we want to use and then negate them all
188 (inspect.isclass(module_cls) and issubclass(module_cls, Module) # is a normal Module class
189 and not issubclass(module_cls, InstancedModule)) # but not a "instantiable" Module
190 or isinstance(module_cls, InstancedModule) # is an already instanced Module
191 ):
192 continue
194 # remember module_instance for default renderer.
195 module_instance = default_module_instance = None
197 for render_name, render in renderers.items(): # look, if a particular renderer should be built
198 # Only continue when module_cls is configured for this render
199 # todo: VIUR4 this is for legacy reasons, can be done better!
200 if not getattr(module_cls, render_name, False):
201 continue
203 # Create a new module instance
204 module_instance = module_cls(
205 module_name, ("/" + render_name if render_name != default else "") + "/" + module_name
206 )
208 # Attach the module-specific or the default render
209 if render_name == default: # default or render (sub)namespace?
210 default_module_instance = module_instance
211 target = resolver
212 else:
213 if getattr(index, render_name, True) is True:
214 # Render is not build yet, or it is just the simple marker that a given render should be build
215 setattr(index, render_name, Module(render_name, "/" + render_name))
217 # Attach the module to the given renderer node
218 setattr(getattr(index, render_name), module_name, module_instance)
219 target = resolver.setdefault(render_name, {})
221 module_instance.register(target, render.get(module_name, render["default"])(parent=module_instance))
223 # Apply Renderers postProcess Filters
224 if "_postProcessAppObj" in render: # todo: This is ugly!
225 render["_postProcessAppObj"](target)
227 # Ugly solution, but there is no better way to do it in ViUR 3:
228 # Allow that any module can be accessed by `conf.main_app.<modulename>`,
229 # either with default render or the last created render.
230 # This behavior does NOT influence the routing.
231 if default_module_instance or module_instance:
232 setattr(index, module_name, default_module_instance or module_instance)
234 # fixme: Below is also ugly...
235 if default in renderers and hasattr(renderers[default]["default"], "renderEmail"):
236 conf.emailRenderer = renderers[default]["default"]().renderEmail
237 elif "html" in renderers:
238 conf.emailRenderer = renderers["html"]["default"]().renderEmail
240 # This might be useful for debugging, please keep it for now.
241 if conf.debug.trace:
242 import pprint
243 logging.debug(pprint.pformat(resolver))
245 conf.main_resolver = resolver
246 conf.main_app = index
249def setup(modules: ModuleType | object, render: ModuleType | object = None, default: str = "html"):
250 """
251 Define whats going to be served by this instance.
253 :param modules: Usually the module provided as *modules* directory within the application.
254 :param render: Usually the module *viur.core.renders*, or a dictionary renderName => renderClass.
255 :param default: Name of the renderer, which will form the root of the application.\
256 This will be the renderer, which wont get a prefix, usually html. \
257 (=> /user instead of /html/user)
258 """
259 # The App Engine SDK patches the threading APIs in `wrap_wsgi_app()` and thus relies on it running before
260 # anything starts a thread. `setup()` breaks that order: it calls `wrap_wsgi_app()` only at its very end,
261 # after `runStartupTasks()`, whose App Engine API calls may already start threads of the API thread pool.
262 # The patch reloads `threading` and `concurrent.futures.thread`; threads started before that drop out of
263 # their registries, are never woken up on shutdown and block the interpreter's exit.
264 # So patch first; the call in `wrap_wsgi_app()` is then a no-op.
265 gae_initialize.InitializeThreadingApis()
267 from viur.core.bones.base import setSystemInitialized
268 # noinspection PyUnresolvedReferences
269 import skeletons # This import is not used here but _must_ remain to ensure that the
270 # application's data models are explicitly imported at some place!
271 for application_id in conf.valid_application_ids:
272 if fnmatch.fnmatch(conf.instance.project_id, application_id):
273 break
274 else:
275 raise RuntimeError(
276 f"""Refusing to start, {conf.instance.project_id=} is not in {conf.valid_application_ids=}""")
277 if not render:
278 import viur.core.render
279 render = viur.core.render
281 __build_app(modules, render, default)
283 # Send warning email in case trace is activated in a cloud environment
284 if ((conf.debug.trace
285 or conf.debug.trace_external_call_routing
286 or conf.debug.trace_internal_call_routing)
287 and (not conf.instance.is_dev_server or conf.debug.dev_server_cloud_logging)):
288 from viur.core import email
289 try:
290 email.send_email_to_admins(
291 "Debug mode enabled",
292 "ViUR just started a new Instance with call tracing enabled! This might log sensitive information!"
293 )
294 except Exception as exc: # OverQuota, whatever
295 logging.exception(exc)
296 # Ensure that our Content Security Policy Header Cache gets build
297 from viur.core import securityheaders
298 securityheaders._rebuildCspHeaderCache()
299 securityheaders._rebuildPermissionHeaderCache()
300 setSystemInitialized()
301 # Assert that all security related headers are in a sane state
302 if conf.security.content_security_policy and conf.security.content_security_policy["_headerCache"]:
303 for k in conf.security.content_security_policy["_headerCache"]:
304 if not k.startswith("Content-Security-Policy"):
305 raise AssertionError("Got unexpected header in "
306 "conf.security.content_security_policy['_headerCache']")
307 if conf.security.strict_transport_security:
308 if not conf.security.strict_transport_security.startswith("max-age"):
309 raise AssertionError("Got unexpected header in conf.security.strict_transport_security")
310 securityheaders._validate_reporting_config()
311 crossDomainPolicies = {None, "none", "master-only", "by-content-type", "all"}
312 if conf.security.x_permitted_cross_domain_policies not in crossDomainPolicies:
313 raise AssertionError("conf.security.x_permitted_cross_domain_policies "
314 f"must be one of {crossDomainPolicies!r}")
315 if conf.security.x_frame_options is not None and isinstance(conf.security.x_frame_options, tuple):
316 mode, uri = conf.security.x_frame_options
317 assert mode in ["deny", "sameorigin", "allow-from"]
318 if mode == "allow-from":
319 assert uri is not None and (uri.lower().startswith("https://") or uri.lower().startswith("http://"))
320 runStartupTasks() # Add a deferred call to run all queued startup tasks
321 i18n.initializeTranslations()
322 if conf.file_hmac_key is None:
323 from viur.core import db
324 key = db.Key("viur-conf", "viur-conf")
325 if not (obj := db.get(key)): # create a new "viur-conf"?
326 logging.info("Creating new viur-conf")
327 obj = db.Entity(key)
329 if "hmacKey" not in obj: # create a new hmacKey
330 logging.info("Creating new hmacKey")
331 obj["hmacKey"] = utils.string.random(length=20)
332 db.put(obj)
334 conf.file_hmac_key = bytes(obj["hmacKey"], "utf-8")
336 if conf.instance.is_dev_server:
337 WIDTH = 80 # defines the standard width
338 FILL = "#" # define sthe fill char (must be len(1)!)
339 PYTHON_VERSION = (sys.version_info.major, sys.version_info.minor, sys.version_info.micro)
341 # extra banner lines for a non-default datastore target
342 datastore_lines = []
343 if conf.db.name:
344 datastore_lines.append(f"""database = \033[1;33m{conf.db.name}\033[0m""")
345 if conf.db.namespace:
346 datastore_lines.append(f"""namespace = \033[1;33m{conf.db.namespace}\033[0m""")
348 # define lines to show
349 lines = (
350 " LOCAL DEVELOPMENT SERVER IS UP AND RUNNING ", # title line
351 f"""project = \033[1;31m{conf.instance.project_id}\033[0m""",
352 f"""python = \033[1;32m{".".join((str(i) for i in PYTHON_VERSION))}\033[0m""",
353 f"""viur = \033[1;32m{".".join((str(i) for i in conf.version))}\033[0m""",
354 *datastore_lines, # only when a non-default db/namespace is set
355 "" # empty line
356 )
358 # first and last line are shown with a cool line made of FILL
359 first_last = (0, len(lines) - 1)
361 # dump to console
362 for i, line in enumerate(lines):
363 print(
364 f"""\033[0m{FILL}{line:{
365 FILL if i in first_last else " "}^{(WIDTH - 2) + (11 if i not in first_last else 0)
366 }}{FILL}"""
367 )
369 return wrap_wsgi_app(app)
372def app(environ: dict, start_response: t.Callable):
373 return request.Router(environ).response(environ, start_response)
376# DEPRECATED ATTRIBUTES HANDLING
378__DEPRECATED_DECORATORS = {
379 # stuff prior viur-core < 3.5
380 "forcePost": ("force_post", force_post),
381 "forceSSL": ("force_ssl", force_ssl),
382 "internalExposed": ("internal_exposed", internal_exposed)
383}
386def __getattr__(attr: str) -> object:
387 if entry := __DEPRECATED_DECORATORS.get(attr): 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 func = entry[1]
389 msg = f"@{attr} was replaced by @{entry[0]}"
390 warnings.warn(msg, DeprecationWarning, stacklevel=2)
391 logging.warning(msg, stacklevel=2)
392 return func
394 return super(__import__(__name__).__class__).__getattr__(attr)