fix(gt-i18n): keep translations loaded after the cache expires

.sampo/changesets/keep-translations-loaded.md

@@ -0,0 +1,5 @@
+---
+pypi/gt-i18n: patch
+---
+
+fix: keep translations loaded after the 60 s cache expiry (serve the cached copy while one background reload runs, keep it when a reload fails), default the CDN URL to cdn.gtx.dev and add the pinned `version_id` and `branch_id` to CDN requests, treat an empty `cache_url` as no CDN loading, and require generaltranslation 0.1.0 and generaltranslation-supported-locales 0.0.6 or later

packages/gt-i18n/pyproject.toml

@@ -10,8 +10,8 @@ requires-python = ">=3.10"
 license = "FSL-1.1-ALv2"
 license-files = ["LICENSE.md"]
 dependencies = [
-    "generaltranslation",
-    "generaltranslation-supported-locales",
+    "generaltranslation>=0.1.0",
+    "generaltranslation-supported-locales>=0.0.6",
     "httpx>=0.25.0",
 ]
 

packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py

@@ -24,12 +24,15 @@ class I18nManager:
         default_locale: The source/default locale.
         locales: List of target locales the application supports.
         project_id: GT project ID (used for CDN loader).
-        cache_url: CDN base URL override.
+        cache_url: CDN base URL override. ``None`` uses the General
+            Translation CDN; an empty string turns CDN loading off.
         store_adapter: Custom storage adapter. Defaults to
             :class:`ContextVarStorageAdapter`.
         load_translations: Custom translation loader. Overrides the
             remote CDN loader when provided.
         cache_expiry_time: Cache expiry in milliseconds.
+        version_id: Pinned translation version for CDN loading.
+        branch_id: General Translation branch of the pinned version.
     """
 
     def __init__(
@@ -44,6 +47,7 @@ def __init__(
         load_translations: TranslationsLoader | None = None,
         cache_expiry_time: int = 60_000,
         version_id: str | None = None,
+        branch_id: str | None = None,
     ) -> None:
         self._version_id = version_id
         self._default_locale = default_locale
@@ -59,12 +63,12 @@ def __init__(
         # Translation loading
         if load_translations is not None:
             loader = load_translations
-        elif project_id:
+        elif project_id and cache_url != "":
             from gt_i18n.i18n_manager._remote_loader import (
                 create_remote_translation_loader,
             )
 
-            loader = create_remote_translation_loader(project_id, cache_url or "")
+            loader = create_remote_translation_loader(project_id, cache_url, version_id=version_id, branch_id=branch_id)
         else:
             loader = lambda locale: {}  # noqa: E731
 
@@ -114,12 +118,22 @@ async def get_translations(self, locale: str | None = None) -> dict[str, str]:
         target = locale or self.get_locale()
         return await self._translations.get_translations(target)
 
+    def get_translations_blocking(self, locale: str | None = None) -> dict[str, str]:
+        """Get translations for a locale (sync; blocks on the first load, reloads expired ones in the background)."""
+        target = locale or self.get_locale()
+        return self._translations.get_translations_blocking(target)
+
     def get_translations_sync(self, locale: str | None = None) -> dict[str, str]:
-        """Get cached translations (sync, returns empty if not loaded)."""
+        """Get the last loaded translations (sync, never loads; empty if never loaded)."""
         target = locale or self.get_locale()
         return self._translations.get_translations_sync(target)
 
     async def load_all_translations(self) -> None:
         """Eagerly load translations for all configured locales."""
         if self._locales:
             await self._translations.load_all(self._locales)
+
+    def load_all_translations_blocking(self) -> None:
+        """Eagerly load translations for all configured locales from synchronous code."""
+        if self._locales:
+            self._translations.load_all_blocking(self._locales)

packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py

@@ -12,29 +12,37 @@
 
 def create_remote_translation_loader(
     project_id: str,
-    cache_url: str = DEFAULT_CACHE_URL,
+    cache_url: str | None = None,
+    *,
+    version_id: str | None = None,
+    branch_id: str | None = None,
 ) -> TranslationsLoader:
     """Create a loader that fetches translations from the CDN.
 
     Args:
         project_id: The GT project ID.
-        cache_url: CDN base URL.
+        cache_url: CDN base URL. Defaults to the General Translation CDN.
+        version_id: Pins the translations to one published version.
+        branch_id: The General Translation branch the version belongs to.
 
     Returns:
-        An async callable ``(locale) -> dict[str, str]``.
+        An async callable ``(locale) -> dict[str, str]`` that raises when the
+        request fails or returns no translations, so the cache keeps its
+        previous translations. The CDN answers some server-side failures with
+        an empty object, which is why empty counts as a failure here.
     """
+    base = f"{cache_url or DEFAULT_CACHE_URL}/{project_id}"
+    version = f"/{version_id}" if version_id else ""
+    params = {"branchId": branch_id} if branch_id else None
 
     async def loader(locale: str) -> dict[str, str]:
-        url = f"{cache_url}/{project_id}/{locale}"
-        try:
-            async with httpx.AsyncClient() as client:
-                resp = await client.get(url, timeout=10.0)
-                if resp.status_code == 200:
-                    data = resp.json()
-                    if isinstance(data, dict):
-                        return data
-        except Exception:
-            pass
-        return {}
+        url = f"{base}/{locale}{version}"
+        async with httpx.AsyncClient() as client:
+            resp = await client.get(url, params=params, timeout=10.0)
+        resp.raise_for_status()
+        data = resp.json()
+        if not isinstance(data, dict) or not data:
+            raise ValueError(f"No translations in the response from {url}")
+        return data
 
     return loader

packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py

@@ -4,8 +4,11 @@
 
 import asyncio
 import inspect
+import threading
 import time
 from collections.abc import Awaitable, Callable
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
 
 TranslationsLoader = Callable[[str], dict[str, str] | Awaitable[dict[str, str]]]
 
@@ -18,58 +21,197 @@ def __init__(self, translations: dict[str, str], loaded_at: float) -> None:
         self.loaded_at = loaded_at
 
 
+async def _await(awaitable: Awaitable[Any]) -> Any:
+    return await awaitable
+
+
+def _run_blocking(awaitable: Awaitable[Any]) -> Any:
+    """Run *awaitable* to completion from synchronous code."""
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(_await(awaitable))
+    # A loop is already running in this thread, so run it on a fresh one in another thread.
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        return pool.submit(asyncio.run, _await(awaitable)).result()
+
+
+def _is_async(loader: TranslationsLoader) -> bool:
+    """Whether *loader* is an ``async def`` function or an object whose ``__call__`` is one."""
+    return inspect.iscoroutinefunction(loader) or inspect.iscoroutinefunction(getattr(loader, "__call__", None))
+
+
 class TranslationsManager:
-    """Caches ``dict[str, str]`` (hash -> translated string) per locale."""
+    """Caches ``dict[str, str]`` (hash -> translated string) per locale.
+
+    An entry expires ``cache_expiry_time`` ms after it loads. An expired entry
+    keeps being served while one reload runs in the background, and a reload
+    that fails keeps the previous translations, so a slow or failing loader
+    never drops a loaded locale back to the source text. Only a locale's first
+    load makes its callers wait.
+
+    From async code, an async loader runs on the caller's event loop, so a
+    loader that holds a client bound to that loop keeps working, and its loads
+    of a locale on that loop share one task. If the loop closes before a load
+    finishes, as a one-off ``asyncio.run()`` does, the load finishes on a
+    thread. Every other load runs on a thread and holds the locale's lock. In
+    both cases, a reload and ``load_all`` for the same locale share one load.
+    """
 
     def __init__(
         self,
         loader: TranslationsLoader,
         cache_expiry_time: int = 60_000,
     ) -> None:
         self._loader = loader
+        self._loader_is_async = _is_async(loader)
         self._cache_expiry_ms = cache_expiry_time
         self._cache: dict[str, _CacheEntry] = {}
-        self._loading: dict[str, asyncio.Task[dict[str, str]]] = {}
+        self._loading: dict[str, asyncio.Task[None]] = {}
+        self._locks: dict[str, threading.Lock] = {}
+        self._locks_guard = threading.Lock()
 
     def _is_expired(self, entry: _CacheEntry) -> bool:
         elapsed_ms = (time.monotonic() - entry.loaded_at) * 1000
         return elapsed_ms > self._cache_expiry_ms
 
-    async def _call_loader(self, locale: str) -> dict[str, str]:
-        result = self._loader(locale)
-        if inspect.isawaitable(result):
-            return await result
-        return result  # type: ignore[return-value]
+    def _fresh(self, locale: str) -> dict[str, str] | None:
+        entry = self._cache.get(locale)
+        if entry is not None and not self._is_expired(entry):
+            return entry.translations
+        return None
+
+    def _store(self, locale: str, translations: dict[str, str] | None) -> dict[str, str]:
+        """Cache a load result; ``None`` is a failed load, which keeps the previous translations."""
+        if translations is None:
+            previous = self._cache.get(locale)
+            translations = previous.translations if previous is not None else {}
+        self._cache[locale] = _CacheEntry(translations, time.monotonic())
+        return translations
+
+    # Async path (FastAPI and other asyncio code)
+
+    async def _load_async(self, locale: str) -> dict[str, str] | None:
+        try:
+            result = self._loader(locale)
+            return await result if inspect.isawaitable(result) else result
+        except Exception:
+            return None
+
+    async def _load_and_store(self, locale: str) -> None:
+        if self._loader_is_async:
+            self._store(locale, await self._load_async(locale))
+        else:
+            # A synchronous loader runs on a worker thread, off the event loop.
+            await asyncio.to_thread(self._load_locked, locale)
+
+    def _start_load(self, locale: str) -> asyncio.Task[None]:
+        """Start a load for *locale* on the running loop, or join the one already running there."""
+        loop = asyncio.get_running_loop()
+        task = self._loading.get(locale)
+        if task is None or task.done() or task.get_loop() is not loop:
+            task = loop.create_task(self._load_and_store(locale))
+            self._loading[locale] = task
+            task.add_done_callback(lambda done: self._load_finished(locale, done))
+        return task
+
+    def _load_finished(self, locale: str, task: asyncio.Task[None]) -> None:
+        if self._loading.get(locale) is task:
+            del self._loading[locale]
+        if task.cancelled():
+            # The loop closed before the load finished, so it finishes on a thread.
+            self._refresh_in_background(locale)
 
     async def get_translations(self, locale: str) -> dict[str, str]:
-        """Load translations for *locale*, using cache when valid."""
+        """Return translations for *locale*, loading them if missing and reloading in the background if expired."""
+        fresh = self._fresh(locale)
+        if fresh is not None:
+            return fresh
         entry = self._cache.get(locale)
-        if entry is not None and not self._is_expired(entry):
+        if entry is not None:
+            if not self._loader_is_async:
+                # The reload runs on a thread, so it outlives this call's event loop.
+                self._refresh_in_background(locale)
+            elif not self._lock_for(locale).locked():  # unless a thread is reloading it already
+                # An async loader reloads on this loop, which it may hold a client for.
+                self._start_load(locale)
             return entry.translations
+        # Shielded, so a cancelled caller does not cancel the load for the others.
+        await asyncio.shield(self._start_load(locale))
+        return self.get_translations_sync(locale)
+
+    async def load_all(self, locales: list[str]) -> None:
+        """Eagerly fetch translations for all *locales*, waiting for every load."""
+        tasks = [self._start_load(loc) for loc in locales if self._fresh(loc) is None]
+        if tasks:
+            await asyncio.gather(*(asyncio.shield(task) for task in tasks))
+
+    # Blocking path (Flask and other synchronous code)
 
-        # Deduplicate concurrent loads
-        if locale in self._loading:
-            return await self._loading[locale]
+    def _lock_for(self, locale: str) -> threading.Lock:
+        with self._locks_guard:
+            return self._locks.setdefault(locale, threading.Lock())
 
-        task = asyncio.ensure_future(self._call_loader(locale))
-        self._loading[locale] = task
+    def _load_blocking(self, locale: str) -> dict[str, str] | None:
         try:
-            translations = await task
+            result = self._loader(locale)
+            return _run_blocking(result) if inspect.isawaitable(result) else result
         except Exception:
-            translations = {}
-        finally:
-            self._loading.pop(locale, None)
+            return None
 
-        self._cache[locale] = _CacheEntry(translations, time.monotonic())
-        return translations
+    def _load_locked(self, locale: str) -> None:
+        """Load *locale* on this thread unless it is fresh, after any load of it already running on a thread."""
+        with self._lock_for(locale):
+            if self._fresh(locale) is None:
+                self._store(locale, self._load_blocking(locale))
 
-    def get_translations_sync(self, locale: str) -> dict[str, str]:
-        """Return cached translations or empty dict (never blocks)."""
+    def _reload_in_background(self, locale: str, lock: threading.Lock) -> None:
+        try:
+            self._store(locale, self._load_blocking(locale))
+        finally:
+            lock.release()
+
+    def _refresh_in_background(self, locale: str) -> None:
+        """Load *locale* on a daemon thread, unless a thread is loading it already."""
+        lock = self._lock_for(locale)
+        if not lock.acquire(blocking=False):
+            return
+        if self._fresh(locale) is not None:  # another reload finished in the meantime
+            lock.release()
+            return
+        try:
+            threading.Thread(target=self._reload_in_background, args=(locale, lock), daemon=True).start()
+        except BaseException:
+            lock.release()
+            raise
+
+    def get_translations_blocking(self, locale: str) -> dict[str, str]:
+        """Return translations for *locale* from synchronous code.
+
+        The first load blocks the callers for that locale. After that, an
+        expired entry is returned at once while one background thread reloads it.
+        """
+        fresh = self._fresh(locale)
+        if fresh is not None:
+            return fresh
         entry = self._cache.get(locale)
-        if entry is not None and not self._is_expired(entry):
+        if entry is not None:
+            self._refresh_in_background(locale)
             return entry.translations
-        return {}
+        with self._lock_for(locale):
+            entry = self._cache.get(locale)
+            if entry is not None:
+                return entry.translations
+            return self._store(locale, self._load_blocking(locale))
+
+    def load_all_blocking(self, locales: list[str]) -> None:
+        """Eagerly fetch translations for all *locales* from synchronous code, waiting for every load."""
+        stale = [loc for loc in locales if self._fresh(loc) is None]
+        if stale:
+            with ThreadPoolExecutor() as pool:
+                list(pool.map(self._load_locked, stale))
 
-    async def load_all(self, locales: list[str]) -> None:
-        """Eagerly fetch translations for all *locales*."""
-        await asyncio.gather(*(self.get_translations(loc) for loc in locales))
+    def get_translations_sync(self, locale: str) -> dict[str, str]:
+        """Return the last loaded translations, even if expired, or an empty dict (never blocks)."""
+        entry = self._cache.get(locale)
+        return entry.translations if entry is not None else {}

packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py

@@ -17,17 +17,19 @@
     "runtimeUrl": "runtime_url",
     "customMapping": "custom_mapping",
     "_versionId": "version_id",
+    "_branchId": "branch_id",
 }
 
 
 class GTConfig(TypedDict, total=False):
     default_locale: str
     locales: list[str]
     project_id: str
-    cache_url: str
+    cache_url: str | None
     runtime_url: str
     custom_mapping: CustomMapping
     version_id: str
+    branch_id: str
 
 
 def load_gt_config(config_path: str | None = None) -> GTConfig: