wip

examples/i18n/python/local/i18n/__init__.py

@@ -3,13 +3,11 @@
 #  (c) Technische Universität Berlin, innoCampus <info@isis.tu-berlin.de>
 from questionpy import Attempt, NeedsManualScoringError, Question, make_question_type_init
 from questionpy.form import FormModel, static_text
-from questionpy.i18n import gettext as _
+from questionpy import i18n
 from questionpy_common.elements import StaticTextElement
 
 
-def _(a):
-    return a
-
+_, _N = i18n.get_for(__package__)
 
 class I18NModel(FormModel):
     txt: StaticTextElement = static_text(

questionpy/_ui.py

@@ -102,9 +102,9 @@ def create_jinja2_environment(attempt: "Attempt", question: "Question") -> jinja
         "question_type": type(question),
     })
 
-    if i18n.is_initialized():
+    translations = i18n.get_translations_of_package(package)
+    if translations:
         env.add_extension("jinja2.ext.i18n")
-        _, translations = i18n.get_state()
-        env.install_gettext_translations(translations, newstyle=True)
+        env.install_gettext_translations(translations, newstyle=True)  # type: ignore[attr-defined]
 
     return env

questionpy/i18n.py

@@ -1,21 +1,44 @@
 import logging
+from collections.abc import Callable
 from contextvars import ContextVar
+from dataclasses import dataclass
 from gettext import GNUTranslations, NullTranslations
 from importlib.resources.abc import Traversable
-
-from questionpy_common.environment import Environment, Package, RequestUser
+from typing import TypeAlias
+
+from questionpy_common.environment import (
+    Package,
+    PackageNamespaceAndShortName,
+    RequestUser,
+    get_qpy_environment,
+)
 from questionpy_common.manifest import SourceManifest
 
 DEFAULT_CATEGORY = "LC_MESSAGES"
 _NULL_TRANSLATIONS = NullTranslations()
 
-_i18n_state: ContextVar[tuple[str, NullTranslations]] = ContextVar(f"_i18n_state")
+
+@dataclass
+class _RequestState:
+    user: RequestUser
+    primary_lang: str
+    translations: NullTranslations
+
+
+@dataclass
+class _DomainState:
+    untranslated_lang: str
+    available_mos: dict[str, Traversable]
+    request_state: _RequestState | None = None
+
+
+_i18n_state: ContextVar[dict[str, _DomainState]] = ContextVar("_i18n_state")
 
 _log = logging.getLogger(__name__)
 
 
-def domain_of(manifest: SourceManifest) -> str:
-    return f"{manifest.namespace}.{manifest.short_name}"
+def domain_of(package: SourceManifest | PackageNamespaceAndShortName) -> str:
+    return f"{package.namespace}.{package.short_name}"
 
 
 def _guess_untranslated_language(package: Package) -> str:
@@ -57,69 +80,105 @@ def _get_available_mos(package: Package) -> dict[str, Traversable]:
     return result
 
 
-def get_state() -> tuple[str, NullTranslations]:
-    """Get the current i18n state of the worker."""
-    state = _i18n_state.get(None)
-    if not state:
-        msg = f"i18n was not initialized. Call {__name__}.{initialize.__name__} first."
+def _require_request_user() -> RequestUser:
+    env = get_qpy_environment()
+    if not env.request_user:
+        msg = "No request is currently being processed."
         raise RuntimeError(msg)
-    return state
-
 
-def is_initialized() -> bool:
-    return _i18n_state.get(None) is not None
+    return env.request_user
 
 
-def initialize(package: Package, env: Environment) -> None:
-    # PLW0603 discourages the global statement, but the alternative would be much less readable.
-    # ruff: noqa: PLW0603
+def get_translations_of_package(package: Package) -> NullTranslations:
+    """Get the current i18n state of the worker."""
+    domain = domain_of(package.manifest)
+    request_state = _ensure_initialized_for_request(domain, package, _require_request_user())
+    return request_state.translations
 
-    if is_initialized():
-        # Prevent multiple initializations, which would add multiple on_request_callbacks overwriting each other.
-        return
 
-    untranslated_lang = _guess_untranslated_language(package)
-    _NULL_TRANSLATIONS.install()
-    _i18n_state.set((untranslated_lang, _NULL_TRANSLATIONS))
+_GettextFun: TypeAlias = Callable[[str], str]
 
-    available_mos = _get_available_mos(package)
 
-    if not available_mos:
-        # We'd never translate anything anyway, prepare_i18n would install the NullTranslations every time.
-        # (Ohne MOs nix los)
-        _log.debug(
-            "No MO files found, messages will not be translated. We'll assume the untranslated strings to be in '%s'.",
-            untranslated_lang,
+def _get_package_owning_module(module_name: str) -> Package:
+    # TODO: Dedupe when #152 is in dev.
+    try:
+        namespace, short_name, *_ = module_name.split(".", maxsplit=2)
+        env = get_qpy_environment()
+        key = PackageNamespaceAndShortName(namespace=namespace, short_name=short_name)
+        return env.packages[key]
+    except (KeyError, ValueError) as e:
+        msg = (
+            "Current package namespace and shortname could not be determined from '__module__' attribute. Please do "
+            "not modify the '__module__' attribute."
         )
-        return
+        raise ValueError(msg) from e
 
-    _log.debug("Found MO files for the following languages: %s", ", ".join(available_mos.keys()))
 
-    def prepare_i18n(request_user: RequestUser) -> None:
-        langs_to_use = [lang for lang in request_user.preferred_languages if lang in available_mos]
+def _ensure_initialized(domain: str, package: Package) -> _DomainState:
+    state_dict = _i18n_state.get(None)
+    if state_dict is None:
+        state_dict = {}
+        _i18n_state.set(state_dict)
 
-        if langs_to_use:
-            _log.debug("Using the following languages for this request: %s", langs_to_use)
-            primary_lang = langs_to_use[0]
+    domain_state = state_dict.get(domain)
+    if not domain_state:
+        untranslated_lang = _guess_untranslated_language(package)
+        available_mos = _get_available_mos(package)
+        if available_mos:
+            _log.debug(
+                "For domain '%s', MO files for the following languages were found: %s",
+                domain,
+                ", ".join(available_mos.keys()),
+            )
         else:
             _log.debug(
-                "There are no MO files for any of the user's preferred languages. Messages will not be translated "
-                "and we'll assume the untranslated strings to be in '%s'.",
+                "For domain '%s', no MO files were found. Messages will not be translated. We'll assume the "
+                "untranslated strings to be in '%s'.",
                 untranslated_lang,
             )
-            primary_lang = untranslated_lang
 
-        translations = _build_translations([available_mos[lang] for lang in langs_to_use])
-        translations.install()
+        domain_state = state_dict[domain] = _DomainState(untranslated_lang, available_mos)
+
+    return domain_state
+
+
+def _ensure_initialized_for_request(domain: str, package: Package, request_user: RequestUser) -> _RequestState:
+    domain_state = _ensure_initialized(domain, package)
+    if domain_state.request_state and domain_state.request_state.user == request_user:
+        return domain_state.request_state
+
+    langs_to_use = [lang for lang in request_user.preferred_languages if lang in domain_state.available_mos]
+
+    if langs_to_use:
+        _log.debug("Using the following languages for this request: %s", langs_to_use)
+        primary_lang = langs_to_use[0]
+    else:
+        _log.debug(
+            "There are no MO files for any of the user's preferred languages. Messages will not be translated "
+            "and we'll assume the untranslated strings to be in '%s'.",
+            domain_state.untranslated_lang,
+        )
+        primary_lang = domain_state.untranslated_lang
+
+    translations = _build_translations([domain_state.available_mos[lang] for lang in langs_to_use])
+    domain_state.request_state = _RequestState(request_user, primary_lang, translations)
+    return domain_state.request_state
+
 
-        _i18n_state.set((primary_lang, translations))
+def get_for(module_name: str) -> tuple[_GettextFun, _GettextFun]:
+    # TODO: Maybe cache this?
+    package = _get_package_owning_module(module_name)
+    domain = domain_of(package.manifest)
+    _ensure_initialized(domain, package)
 
-    env.register_on_request_callback(prepare_i18n)
+    def gettext(message: str) -> str:
+        request_state = _ensure_initialized_for_request(domain, package, _require_request_user())
+        return request_state.translations.gettext(message)
 
+    def ngettext(message: str) -> str:
+        return message
 
-def gettext(msgid: str) -> str:
-    _, translations = get_state()
-    return translations.gettext(msgid)
+    return gettext, ngettext
 
 
-__all__ = ["DEFAULT_CATEGORY", "domain_of", "get_state", "gettext", "initialize", "is_initialized"]
+__all__ = ["DEFAULT_CATEGORY", "domain_of", "get", "get_translations_of_package"]

questionpy_sdk/commands/i18n/_update.py

@@ -1,6 +1,5 @@
 import itertools
 import operator
-from collections.abc import Iterator
 from datetime import datetime, timedelta
 from pathlib import Path
 
@@ -13,8 +12,13 @@
 from questionpy_sdk.package.source import PackageSource
 
 
-def _update_domain(ctx: click.Context, package: PackageSource, domain: str, po_files: list[tuple[str, str, Path]],
-                   pot_file: Path | None = None) -> None:
+def _update_domain(
+    ctx: click.Context,
+    package: PackageSource,
+    domain: str,
+    po_files: list[tuple[str, str, Path]],
+    pot_file: Path | None = None,
+) -> None:
     default_pot_path = package.path / "locale" / f"{domain}.pot"
     if not pot_file:
         pot_file = default_pot_path
@@ -34,7 +38,7 @@ def _update_domain(ctx: click.Context, package: PackageSource, domain: str, po_f
         if now - template_catalog.creation_date > timedelta(hours=1):
             click.echo("Warning: .pot file is more than 1 hour old.")
 
-    for (locale, _, po_file) in po_files:
+    for locale, _, po_file in po_files:
         cmd = babel.messages.frontend.UpdateCatalog()
         cmd.locale = locale
         cmd.input_file = pot_file
@@ -51,8 +55,9 @@ def _update_domain(ctx: click.Context, package: PackageSource, domain: str, po_f
 @click.pass_context
 def update(ctx: click.Context, package_path: Path, pot: Path | None = None, domain: str | None = None) -> None:
     package = PackageSource(package_path)
-    pos_by_domain = {domain: list(pos) for domain, pos in
-                     itertools.groupby(package.discover_po_files(), operator.itemgetter(1))}
+    pos_by_domain = {
+        domain: list(pos) for domain, pos in itertools.groupby(package.discover_po_files(), operator.itemgetter(1))
+    }
 
     if domain:
         if domain not in pos_by_domain:

questionpy_sdk/package/builder.py

@@ -156,8 +156,11 @@ def _compile_pos(self) -> None:
             for locale, domain, po_file in self._source.discover_po_files():
                 if po_file.with_suffix(".mo").exists():
                     # By default, Poedit also saves a compiled .mo file, so we warn the user if one exists.
-                    log.warning("The existing .mo file at '%s' will not be used, '%s' will be compiled instead.",
-                                po_file.with_suffix(".mo"), po_file.name)
+                    log.warning(
+                        "The existing .mo file at '%s' will not be used, '%s' will be compiled instead.",
+                        po_file.with_suffix(".mo"),
+                        po_file.name,
+                    )
 
                 outfile = tempdir / locale / i18n.DEFAULT_CATEGORY / f"{domain}.mo"
                 outfile.parent.mkdir(parents=True, exist_ok=True)