feat(cache): cache game/localization versions from /metadata with configurable TTL

CHANGELOG.md

@@ -1,6 +1,36 @@
 # CHANGELOG
 
 <!-- insertion marker -->
+<a name="v2.2.0"></a>
+
+## [v2.2.0](https://github.com/swgoh-utils/comlink-python/compare/v2.1.0...v2.2.0) (2026-07-12)
+
+### Features
+
+- **helpers:** add localization dictionary parsing with sync and async support (#108) ([45dce8a](https://github.com/swgoh-utils/comlink-python/commit/45dce8a5f402ccb192b9598d308dd86a703ea43b))
+
+### Bug Fixes
+
+- **tests:** switch HMAC rejection tests from GET to POST endpoint (#89) ([75feb67](https://github.com/swgoh-utils/comlink-python/commit/75feb6701648cbaf986624bb90532b8fe24d443a))
+- **helpers:** ensure arena payout time adjusts correctly when shifted to past ([c9e3f59](https://github.com/swgoh-utils/comlink-python/commit/c9e3f59b91a51137867f766f448d7deda07e90b0))
+- **helpers:** handle multi-day offsets in get_arena_payout ([e8a6e05](https://github.com/swgoh-utils/comlink-python/commit/e8a6e05d997559c23734dda5fee2c44cd172d2e8))
+
+<a name="v2.1.0"></a>
+
+## [v2.1.0](https://github.com/swgoh-utils/comlink-python/compare/v2.0.7...v2.1.0) (2026-06-03)
+
+### Bug Fixes
+
+- **examples): update string quoting and rename params for localization bundle calls docs(helpers): document parse_swgoh_string and its extended tag grammar chore: add commitlint config and ignore .pythonrc.py fix(helpers:** extend parse_swgoh_string to cover full NGUI tag set (#83) ([1536853](https://github.com/swgoh-utils/comlink-python/commit/15368533fc09cbbf60964d3ca5044fc9e8c91499))
+
+<a name="v2.0.7"></a>
+
+## [v2.0.7](https://github.com/swgoh-utils/comlink-python/compare/v2.0.6...v2.0.7) (2026-03-30)
+
+### Bug Fixes
+
+- update `sanitize_url` to handle HTTPS URLs without ports, update tests for improved coverage ([577951a](https://github.com/swgoh-utils/comlink-python/commit/577951a6877f47a62c7bf062399fcca6611c9caf))
+
 
 ## [v2.0.6](https://github.com/swgoh-utils/comlink-python/releases/tag/v2.0.6) - 2026-03-29
 

README.md

@@ -214,6 +214,23 @@ Constructor parameters for `SwgohComlink` and `SwgohComlinkAsync`:
 | `port` | `int` | `3000` | Comlink TCP port (used with `host`) |
 | `stats_port` | `int` | `3223` | Stats service TCP port (used with `host`) |
 | `verify_ssl` | `bool` | `True` | Enable TLS certificate verification |
+| `version_cache_ttl` | `float` | `3600` | Seconds to cache game/localization versions from `/metadata` (see below) |
+
+### Version caching
+
+Calls to `/data` and `/localization` require the current `latestGamedataVersion` /
+`latestLocalizationBundleVersion` values from the `/metadata` endpoint. When you call
+`get_game_data()`, `get_localization()`, or `get_latest_game_data_version()` without an explicit
+version, the client fetches these values once and caches them on the instance for
+`version_cache_ttl` seconds (default: 1 hour), instead of making a fresh `/metadata` request every
+time.
+
+- Pass `version_cache_ttl=0` to disable caching (the pre-v2.3 behavior), or `math.inf` to cache
+  for the client's lifetime.
+- Explicit `version=` / `localization_id=` arguments always bypass the cache.
+- Call `invalidate_version_cache()` or `get_latest_game_data_version(refresh=True)` to force a
+  re-fetch, e.g. right after a game update lands.
+- A successful `get_game_metadata()` call also refreshes the cache opportunistically.
 
 ## Available Methods
 
@@ -234,7 +251,8 @@ Methods available on both `SwgohComlink` and `SwgohComlinkAsync` (async methods
 | `get_leaderboard(leaderboard_type, league, division, ...)` | Get GAC leaderboard data |
 | `get_guild_leaderboard(leaderboard_id, count, enums)` | Get guild leaderboard data |
 | `get_unit_stats(request_payload, flags, language)` | Calculate unit stats via swgoh-stats |
-| `get_latest_game_data_version()` | Get latest game data and language versions |
+| `get_latest_game_data_version(refresh)` | Get latest game data and language versions (cached) |
+| `invalidate_version_cache()` | Discard cached versions so the next lookup re-fetches `/metadata` |
 | `get_name_spaces(only_compatible, enums)` | Get available namespaces |
 | `get_segmented_content(content_name_space, accept_language, enums)` | Retrieve segmented content |
 

pyproject.toml

@@ -135,7 +135,6 @@ ignore = [
 [tool.ruff.lint.per-file-ignores]
 # Existing violations to address incrementally — do not add new suppressions
 "src/swgoh_comlink/__init__.py" = ["UP036"]   # outdated version block (sys.exit check)
-"src/swgoh_comlink/exceptions.py" = ["B904"]  # raise-without-from (tracked in issue backlog)
 "src/swgoh_comlink/helpers/*.py" = ["B007", "B904", "SIM102"]  # existing patterns to clean up
 
 [tool.ruff.lint.isort]

src/swgoh_comlink/_base.py

@@ -11,22 +11,56 @@
 import os
 import time
 from collections.abc import Callable
+from dataclasses import dataclass
 from json import dumps
 from typing import Any
 from urllib.parse import urlparse, urlunparse
 
 from typing_extensions import Self
 
-from .exceptions import SwgohComlinkValueError
+from .exceptions import SwgohComlinkException, SwgohComlinkValueError
 from .helpers import Constants, DataItems
 
-__all__ = ["SwgohComlinkBase", "param_alias", "DEFAULT_TIMEOUT", "GAME_DATA_TIMEOUT"]
+__all__ = [
+    "SwgohComlinkBase",
+    "param_alias",
+    "DEFAULT_TIMEOUT",
+    "GAME_DATA_TIMEOUT",
+    "DEFAULT_VERSION_CACHE_TTL",
+]
 
 # Keys whose values must be masked in logs, repr, and debug output.
 _SENSITIVE_KEYS = frozenset({"secret_key", "access_key"})
 
 DEFAULT_TIMEOUT: float = 120.0
 GAME_DATA_TIMEOUT: float = 300.0
+DEFAULT_VERSION_CACHE_TTL: float = 3600.0
+
+# Indirection over the monotonic clock so tests can control cache expiry.
+_now: Callable[[], float] = time.monotonic
+
+
+@dataclass(frozen=True)
+class _CachedVersions:
+    """Game data and localization bundle versions cached from a /metadata response.
+
+    Fields are optional because mocked or unusual /metadata responses may omit
+    either key; accessors raise only when the missing value is actually needed.
+    """
+
+    game: str | None
+    language: str | None
+    expires_at: float
+
+    def require_game(self) -> str:
+        if self.game is None:
+            raise SwgohComlinkException("'latestGamedataVersion' was missing from the /metadata response.")
+        return self.game
+
+    def require_language(self) -> str:
+        if self.language is None:
+            raise SwgohComlinkException("'latestLocalizationBundleVersion' was missing from the /metadata response.")
+        return self.language
 
 
 def param_alias(param: str, alias: str) -> Callable[..., Any]:
@@ -110,6 +144,7 @@ def __init__(
         port: int = 3000,
         stats_port: int = 3223,
         verify_ssl: bool = True,
+        version_cache_ttl: float = DEFAULT_VERSION_CACHE_TTL,
     ):
         from swgoh_comlink import version
 
@@ -118,6 +153,11 @@ def __init__(
         self.stats_url_base = sanitize_url(stats_url)
         self.hmac = False
         self.verify_ssl = verify_ssl
+        # NaN fails the >= comparison, so it is rejected here along with negatives.
+        if not version_cache_ttl >= 0:
+            raise SwgohComlinkValueError("version_cache_ttl must be a non-negative number of seconds.")
+        self.version_cache_ttl = version_cache_ttl
+        self._version_cache: _CachedVersions | None = None
 
         # host and port parameters override defaults
         if host:
@@ -166,6 +206,33 @@ def _construct_request_headers(
             req_headers["Authorization"] = f"HMAC-SHA256 Credential={self.access_key},Signature={hmac_digest}"
         return req_headers
 
+    # ── Version cache ────────────────────────────────────────────────────
+
+    def invalidate_version_cache(self) -> None:
+        """Discard the cached game/localization versions so the next lookup re-fetches /metadata."""
+        self._version_cache = None
+
+    def _cached_versions(self) -> _CachedVersions | None:
+        """Return the cached version entry when caching is enabled and the entry is fresh."""
+        cached = self._version_cache
+        if cached is None or _now() >= cached.expires_at:
+            return None
+        return cached
+
+    def _store_versions(self, game: str | None, language: str | None) -> _CachedVersions:
+        """Build a version entry from *game* and *language* and cache it when caching applies."""
+        entry = _CachedVersions(game=game, language=language, expires_at=_now() + self.version_cache_ttl)
+        if self.version_cache_ttl > 0 and (game is not None or language is not None):
+            self._version_cache = entry
+        return entry
+
+    @staticmethod
+    def _versions_from_metadata(metadata: dict[str, Any]) -> tuple[str | None, str | None]:
+        """Extract the (game, language) version strings from a /metadata response, if present."""
+        game = metadata.get("latestGamedataVersion")
+        language = metadata.get("latestLocalizationBundleVersion")
+        return str(game) if game else None, str(language) if language else None
+
     # ── Static payload builders ──────────────────────────────────────────
 
     @staticmethod

src/swgoh_comlink/exceptions.py

@@ -5,19 +5,10 @@
 
 from __future__ import annotations
 
-import logging
-
-logger = logging.getLogger(__name__)
-
 
 class SwgohComlinkException(Exception):
     """Base class for exceptions in this module."""
 
-    def __init__(self, message: str | Exception) -> None:
-        super().__init__(message)
-        # Log at error level; callers are responsible for traceback context
-        logger.exception(f"SwgohComlinkException: {message}")
-
 
 class SwgohComlinkValueError(SwgohComlinkException, ValueError):
     """Raised when an argument value is invalid."""

src/swgoh_comlink/swgoh_comlink.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import threading
 from json import loads
 from typing import Any, cast
 
@@ -14,6 +15,7 @@
     DEFAULT_TIMEOUT,
     GAME_DATA_TIMEOUT,
     SwgohComlinkBase,
+    _CachedVersions,
     param_alias,
 )
 from .exceptions import SwgohComlinkException, SwgohComlinkValueError
@@ -38,6 +40,8 @@ class SwgohComlink(SwgohComlinkBase):
         port (int): TCP port number where the swgoh-comlink service is running [Default: 3000]
         stats_port (int): TCP port number of where the comlink-stats service is running [Default: 3223]
         verify_ssl (bool): Whether to verify TLS certificates. [Default: True]
+        version_cache_ttl (float): Seconds to cache the game/localization versions fetched from
+            /metadata. 0 disables caching; ``math.inf`` caches for the client lifetime. [Default: 3600]
 
     Notes:
         If the 'host' and 'port' parameters are provided, the 'url' and 'stats_url' parameters are ignored.
@@ -58,6 +62,7 @@ class SwgohComlink(SwgohComlinkBase):
 
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(**kwargs)
+        self._version_lock = threading.Lock()
         self.client = httpx.Client(
             base_url=self.url_base,
             headers={"Content-Type": "application/json"},
@@ -140,9 +145,28 @@ def _post(
         """
         return self._request(method="POST", endpoint=endpoint, payload=payload, stats=stats, timeout=timeout)
 
+    def _get_versions(self, refresh: bool = False) -> _CachedVersions:
+        """Return the current game/localization versions, served from the instance cache.
+
+        Args:
+            refresh: When True, bypass the cache and fetch fresh values from /metadata.
+        """
+        if not refresh:
+            cached = self._cached_versions()
+            if cached is not None:
+                return cached
+        with self._version_lock:
+            # Double-check inside the lock so concurrent cold-cache callers
+            # coalesce into a single /metadata request.
+            if not refresh:
+                cached = self._cached_versions()
+                if cached is not None:
+                    return cached
+            metadata = self.get_game_metadata()
+            return self._store_versions(*self._versions_from_metadata(metadata))
+
     def _get_game_version(self) -> str:
-        md = self.get_game_metadata()
-        return str(md["latestGamedataVersion"])
+        return self._get_versions().require_game()
 
     # ── Public API methods ───────────────────────────────────────────────
 
@@ -253,8 +277,7 @@ def get_localization(
             A dictionary containing the localization data.
         """
         if not localization_id:
-            current_game_version = self.get_latest_game_data_version()
-            localization_id = current_game_version["language"]
+            localization_id = self._get_versions().require_language()
 
         if locale:
             assert localization_id is not None
@@ -280,10 +303,17 @@ def get_game_metadata(self, client_specs: dict[str, Any] | None = None, enums: b
             A dictionary containing the game metadata.
         """
         if client_specs:
-            payload: dict[str, Any] = {"payload": {"client_specs": client_specs}, "enums": enums}
+            payload: dict[str, Any] = {"payload": {"clientSpecs": client_specs}, "enums": enums}
         else:
             payload = {}
-        return cast(dict[str, Any], self._post(endpoint="metadata", payload=payload))
+        metadata = cast(dict[str, Any], self._post(endpoint="metadata", payload=payload))
+        # Opportunistically refresh the version cache; enums=True responses are
+        # skipped since their values may be translated.
+        if not enums and isinstance(metadata, dict):
+            game, language = self._versions_from_metadata(metadata)
+            if game is not None and language is not None:
+                self._store_versions(game, language)
+        return metadata
 
     # alias for non PEP usage of direct endpoint calls
     getGameMetaData = get_game_metadata
@@ -501,17 +531,19 @@ def get_segmented_content(
 
     # ── Helper methods ───────────────────────────────────────────────────
 
-    def get_latest_game_data_version(self) -> dict[str, Any]:
+    def get_latest_game_data_version(self, refresh: bool = False) -> dict[str, Any]:
         """Get the latest game data and language bundle versions.
 
+        Results are served from the instance version cache (see ``version_cache_ttl``).
+
+        Args:
+            refresh: When True, bypass the cache and fetch fresh values from /metadata.
+
         Returns:
             Dictionary with 'game' and 'language' version strings.
         """
-        current_metadata = self.get_metadata()
-        return {
-            "game": current_metadata["latestGamedataVersion"],
-            "language": current_metadata["latestLocalizationBundleVersion"],
-        }
+        versions = self._get_versions(refresh=refresh)
+        return {"game": versions.require_game(), "language": versions.require_language()}
 
     # alias for shorthand call
     getVersion = get_latest_game_data_version