From 7b917dea5ce146c2cce9bbd90b436ba010f01207 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]"
<41898282+github-actions[bot]@users.noreply.github.com>
Date: Sun, 12 Jul 2026 01:52:59 +0000
Subject: [PATCH 1/4] docs(changelog): update for v2.2.0
---
CHANGELOG.md | 30 ++++++++++++++++++++++++++++++
1 file changed, 30 insertions(+)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7daa842..f43ca0e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,36 @@
# CHANGELOG
+
+
+## [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))
+
+
+
+## [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))
+
+
+
+## [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
From f841fa88bd64abc6a66326ee0654e367976733ce Mon Sep 17 00:00:00 2001
From: MarTrepodi
Date: Sun, 12 Jul 2026 17:35:16 -0400
Subject: [PATCH 2/4] fix(exceptions): stop logging tracebacks from exception
constructors
Constructing a SwgohComlinkException logged an error-level traceback via
logger.exception() even for expected, caught validation errors, and calling
it outside an except block emitted 'NoneType: None' tracebacks. Callers own
logging context; the constructor override is removed along with the now-moot
B904 per-file ruff ignore.
---
pyproject.toml | 1 -
src/swgoh_comlink/exceptions.py | 9 ---------
tests/unit/test_exceptions.py | 36 +++++++++++++++++++++++++++++++++
3 files changed, 36 insertions(+), 10 deletions(-)
create mode 100644 tests/unit/test_exceptions.py
diff --git a/pyproject.toml b/pyproject.toml
index df1c4f8..39f8bc0 100644
--- a/pyproject.toml
+++ b/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]
diff --git a/src/swgoh_comlink/exceptions.py b/src/swgoh_comlink/exceptions.py
index d22197a..06d3a65 100644
--- a/src/swgoh_comlink/exceptions.py
+++ b/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."""
diff --git a/tests/unit/test_exceptions.py b/tests/unit/test_exceptions.py
new file mode 100644
index 0000000..1e9cbfe
--- /dev/null
+++ b/tests/unit/test_exceptions.py
@@ -0,0 +1,36 @@
+"""Tests for the swgoh_comlink exception hierarchy."""
+
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+from swgoh_comlink.exceptions import (
+ SwgohComlinkException,
+ SwgohComlinkTypeError,
+ SwgohComlinkValueError,
+)
+
+
+def test_hierarchy():
+ assert issubclass(SwgohComlinkValueError, SwgohComlinkException)
+ assert issubclass(SwgohComlinkValueError, ValueError)
+ assert issubclass(SwgohComlinkTypeError, SwgohComlinkException)
+ assert issubclass(SwgohComlinkTypeError, TypeError)
+
+
+def test_constructing_exceptions_does_not_log(caplog: pytest.LogCaptureFixture):
+ with caplog.at_level(logging.DEBUG):
+ SwgohComlinkException("boom")
+ try:
+ raise SwgohComlinkValueError("bad value")
+ except SwgohComlinkValueError:
+ pass
+
+ assert caplog.records == []
+
+
+def test_exception_message_preserved():
+ exc = SwgohComlinkException("HTTP 500: oops")
+ assert str(exc) == "HTTP 500: oops"
From 1909912fc10b1deb822b816b46a0e7577da7e422 Mon Sep 17 00:00:00 2001
From: MarTrepodi
Date: Sun, 12 Jul 2026 17:35:39 -0400
Subject: [PATCH 3/4] fix(client): send clientSpecs key expected by /metadata
endpoint
The /metadata request schema requires the camelCase 'clientSpecs' key and
declares additionalProperties: false, so the snake_case 'client_specs' key
sent by get_game_metadata() was rejected by the server. Verified against
the comlink OpenAPI spec (v0.40.1).
---
src/swgoh_comlink/swgoh_comlink.py | 2 +-
src/swgoh_comlink/swgoh_comlink_async.py | 2 +-
tests/test_get_metadata.py | 2 +-
tests/unit/test_async_client_mocked.py | 2 +-
tests/unit/test_client_request_mocked.py | 2 +-
5 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/src/swgoh_comlink/swgoh_comlink.py b/src/swgoh_comlink/swgoh_comlink.py
index 0c3fa54..4ecf821 100644
--- a/src/swgoh_comlink/swgoh_comlink.py
+++ b/src/swgoh_comlink/swgoh_comlink.py
@@ -280,7 +280,7 @@ 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))
diff --git a/src/swgoh_comlink/swgoh_comlink_async.py b/src/swgoh_comlink/swgoh_comlink_async.py
index 2c83948..ee00a39 100644
--- a/src/swgoh_comlink/swgoh_comlink_async.py
+++ b/src/swgoh_comlink/swgoh_comlink_async.py
@@ -278,7 +278,7 @@ async def get_game_metadata(
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], await self._post(endpoint="metadata", payload=payload))
diff --git a/tests/test_get_metadata.py b/tests/test_get_metadata.py
index e6438cf..1a6cd43 100644
--- a/tests/test_get_metadata.py
+++ b/tests/test_get_metadata.py
@@ -32,7 +32,7 @@ def test_get_metadata_with_client_specs(self, mock_post):
call_kwargs = mock_post.call_args
payload = call_kwargs.kwargs.get("payload") or call_kwargs[1].get("payload")
- self.assertIn("client_specs", payload["payload"])
+ self.assertIn("clientSpecs", payload["payload"])
if __name__ == "__main__":
diff --git a/tests/unit/test_async_client_mocked.py b/tests/unit/test_async_client_mocked.py
index f0d12a3..5ce2ab1 100644
--- a/tests/unit/test_async_client_mocked.py
+++ b/tests/unit/test_async_client_mocked.py
@@ -257,7 +257,7 @@ async def test_get_game_metadata_with_client_specs(httpx_mock: HTTPXMock):
await client.get_game_metadata(client_specs=specs, enums=True)
body = json.loads(httpx_mock.get_request().content)
- assert body == {"payload": {"client_specs": specs}, "enums": True}
+ assert body == {"payload": {"clientSpecs": specs}, "enums": True}
await client.aclose()
diff --git a/tests/unit/test_client_request_mocked.py b/tests/unit/test_client_request_mocked.py
index c76898c..925b5c7 100644
--- a/tests/unit/test_client_request_mocked.py
+++ b/tests/unit/test_client_request_mocked.py
@@ -230,7 +230,7 @@ def test_get_game_metadata_with_client_specs(httpx_mock: HTTPXMock):
client.get_game_metadata(client_specs=specs, enums=True)
body = json.loads(httpx_mock.get_request().content)
- assert body == {"payload": {"client_specs": specs}, "enums": True}
+ assert body == {"payload": {"clientSpecs": specs}, "enums": True}
# ── Player Arena ─────────────────────────────────────────────────────────
From 8cdd7719619ac7b83ceb66c069825572e17d018a Mon Sep 17 00:00:00 2001
From: MarTrepodi
Date: Sun, 12 Jul 2026 17:35:58 -0400
Subject: [PATCH 4/4] feat(cache): cache game/localization versions from
/metadata with configurable TTL
Version-less get_game_data(), get_localization(), and
get_latest_game_data_version() calls previously made a fresh /metadata
round-trip every time. Both clients now cache latestGamedataVersion and
latestLocalizationBundleVersion on the instance for version_cache_ttl
seconds (default 3600). Cold-cache lookups single-flight behind a
threading.Lock (sync) or asyncio.Lock (async), successful
get_game_metadata() calls refresh the cache opportunistically, and
explicit version arguments always bypass it. Set version_cache_ttl=0 to
restore the previous uncached behavior, math.inf to cache for the client
lifetime, or call invalidate_version_cache() / pass refresh=True to force
a re-fetch.
---
README.md | 20 ++-
src/swgoh_comlink/_base.py | 71 +++++++-
src/swgoh_comlink/swgoh_comlink.py | 54 ++++--
src/swgoh_comlink/swgoh_comlink_async.py | 54 ++++--
tests/unit/test_version_cache.py | 214 +++++++++++++++++++++++
5 files changed, 388 insertions(+), 25 deletions(-)
create mode 100644 tests/unit/test_version_cache.py
diff --git a/README.md b/README.md
index ca9bb35..4758aad 100644
--- a/README.md
+++ b/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 |
diff --git a/src/swgoh_comlink/_base.py b/src/swgoh_comlink/_base.py
index d99c6ff..d6af1fe 100644
--- a/src/swgoh_comlink/_base.py
+++ b/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
diff --git a/src/swgoh_comlink/swgoh_comlink.py b/src/swgoh_comlink/swgoh_comlink.py
index 4ecf821..37bd36c 100644
--- a/src/swgoh_comlink/swgoh_comlink.py
+++ b/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
@@ -283,7 +306,14 @@ def get_game_metadata(self, client_specs: dict[str, Any] | None = None, enums: b
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
diff --git a/src/swgoh_comlink/swgoh_comlink_async.py b/src/swgoh_comlink/swgoh_comlink_async.py
index ee00a39..9cc1341 100644
--- a/src/swgoh_comlink/swgoh_comlink_async.py
+++ b/src/swgoh_comlink/swgoh_comlink_async.py
@@ -5,6 +5,7 @@
from __future__ import annotations
+import asyncio
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 SwgohComlinkAsync(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.
@@ -53,6 +57,7 @@ class SwgohComlinkAsync(SwgohComlinkBase):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
+ self._version_lock = asyncio.Lock()
connection_limits = httpx.Limits(keepalive_expiry=None)
self.client = httpx.AsyncClient(
base_url=self.url_base,
@@ -136,9 +141,28 @@ async def _post(
"""
return await self._request(method="POST", endpoint=endpoint, payload=payload, stats=stats, timeout=timeout)
+ async 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
+ async 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 = await self.get_game_metadata()
+ return self._store_versions(*self._versions_from_metadata(metadata))
+
async def _get_game_version(self) -> str:
- md = await self.get_game_metadata()
- return str(md["latestGamedataVersion"])
+ return (await self._get_versions()).require_game()
# ── Public API methods ───────────────────────────────────────────────
@@ -249,8 +273,7 @@ async def get_localization(
A dictionary containing the localization data.
"""
if not localization_id:
- current_game_version = await self.get_latest_game_data_version()
- localization_id = current_game_version["language"]
+ localization_id = (await self._get_versions()).require_language()
if locale:
assert localization_id is not None
@@ -281,7 +304,14 @@ async def get_game_metadata(
payload: dict[str, Any] = {"payload": {"clientSpecs": client_specs}, "enums": enums}
else:
payload = {}
- return cast(dict[str, Any], await self._post(endpoint="metadata", payload=payload))
+ metadata = cast(dict[str, Any], await 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
@@ -499,17 +529,19 @@ async def get_segmented_content(
# ── Helper methods ───────────────────────────────────────────────────
- async def get_latest_game_data_version(self) -> dict[str, Any]:
+ async 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 = await self.get_metadata()
- return {
- "game": current_metadata["latestGamedataVersion"],
- "language": current_metadata["latestLocalizationBundleVersion"],
- }
+ versions = await self._get_versions(refresh=refresh)
+ return {"game": versions.require_game(), "language": versions.require_language()}
# alias for shorthand call
getVersion = get_latest_game_data_version
diff --git a/tests/unit/test_version_cache.py b/tests/unit/test_version_cache.py
new file mode 100644
index 0000000..7439efa
--- /dev/null
+++ b/tests/unit/test_version_cache.py
@@ -0,0 +1,214 @@
+"""Tests for the instance-level game/localization version cache."""
+
+from __future__ import annotations
+
+import asyncio
+import math
+
+import pytest
+from pytest_httpx import HTTPXMock
+
+from swgoh_comlink import SwgohComlink, SwgohComlinkAsync
+from swgoh_comlink.exceptions import SwgohComlinkException, SwgohComlinkValueError
+
+# Not every test exercises all three mocked endpoints.
+pytestmark = [pytest.mark.httpx_mock(assert_all_responses_were_requested=False)]
+
+BASE_URL = "http://localhost:3000"
+METADATA = {
+ "latestGamedataVersion": "game-v1",
+ "latestLocalizationBundleVersion": "lang-v1",
+}
+
+
+def _metadata_requests(httpx_mock: HTTPXMock) -> list:
+ return [r for r in httpx_mock.get_requests() if r.url.path == "/metadata"]
+
+
+def _mock_endpoints(httpx_mock: HTTPXMock, metadata: dict | None = None) -> None:
+ httpx_mock.add_response(url=f"{BASE_URL}/metadata", json=metadata or METADATA, is_reusable=True)
+ httpx_mock.add_response(url=f"{BASE_URL}/data", json={"units": []}, is_reusable=True)
+ httpx_mock.add_response(url=f"{BASE_URL}/localization", json={"localizationBundle": ""}, is_reusable=True)
+
+
+@pytest.fixture
+def clock(monkeypatch: pytest.MonkeyPatch) -> dict[str, float]:
+ """Controllable monotonic clock for cache expiry."""
+ state = {"t": 1000.0}
+ monkeypatch.setattr("swgoh_comlink._base._now", lambda: state["t"])
+ return state
+
+
+# ── Sync client ──────────────────────────────────────────────────────────
+
+
+def test_repeated_game_data_calls_hit_metadata_once(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL)
+
+ client.get_game_data()
+ client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+
+def test_cache_shared_between_game_data_and_localization(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL)
+
+ client.get_game_data()
+ client.get_localization()
+
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+
+def test_ttl_expiry_triggers_refetch(httpx_mock: HTTPXMock, clock: dict[str, float]):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL, version_cache_ttl=60.0)
+
+ client.get_game_data()
+ clock["t"] += 59.0
+ client.get_game_data()
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+ clock["t"] += 2.0 # past the 60s TTL
+ client.get_game_data()
+ assert len(_metadata_requests(httpx_mock)) == 2
+
+
+def test_ttl_zero_disables_caching(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL, version_cache_ttl=0)
+
+ client.get_game_data()
+ client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 2
+
+
+def test_infinite_ttl_caches_forever(httpx_mock: HTTPXMock, clock: dict[str, float]):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL, version_cache_ttl=math.inf)
+
+ client.get_game_data()
+ clock["t"] += 10**9
+ client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+
+def test_invalidate_version_cache_forces_refetch(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL)
+
+ client.get_game_data()
+ client.invalidate_version_cache()
+ client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 2
+
+
+def test_get_latest_game_data_version_refresh_bypasses_cache(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL)
+
+ versions = client.get_latest_game_data_version()
+ assert versions == {"game": "game-v1", "language": "lang-v1"}
+ assert client.get_latest_game_data_version() == versions
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+ client.get_latest_game_data_version(refresh=True)
+ assert len(_metadata_requests(httpx_mock)) == 2
+
+
+def test_get_game_metadata_populates_cache(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL)
+
+ client.get_game_metadata()
+ client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+
+def test_get_game_metadata_with_enums_does_not_populate_cache(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL)
+
+ client.get_game_metadata(enums=True)
+ client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 2
+
+
+def test_explicit_version_skips_metadata_entirely(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ client = SwgohComlink(url=BASE_URL)
+
+ client.get_game_data(version="explicit-version")
+ client.get_localization(localization_id="explicit-loc-id")
+
+ assert len(_metadata_requests(httpx_mock)) == 0
+
+
+def test_partial_metadata_supports_game_data_only(httpx_mock: HTTPXMock):
+ """A /metadata response missing the localization key still serves get_game_data."""
+ _mock_endpoints(httpx_mock, metadata={"latestGamedataVersion": "game-only"})
+ client = SwgohComlink(url=BASE_URL)
+
+ client.get_game_data()
+
+ with pytest.raises(SwgohComlinkException, match="latestLocalizationBundleVersion"):
+ client.get_localization()
+
+
+def test_negative_ttl_rejected():
+ with pytest.raises(SwgohComlinkValueError):
+ SwgohComlink(url=BASE_URL, version_cache_ttl=-1)
+
+
+# ── Async client ─────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_async_repeated_game_data_calls_hit_metadata_once(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ async with SwgohComlinkAsync(url=BASE_URL) as client:
+ await client.get_game_data()
+ await client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+
+@pytest.mark.asyncio
+async def test_async_concurrent_cold_calls_single_flight(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ async with SwgohComlinkAsync(url=BASE_URL) as client:
+ await asyncio.gather(client.get_game_data(), client.get_game_data(), client.get_localization())
+
+ assert len(_metadata_requests(httpx_mock)) == 1
+
+
+@pytest.mark.asyncio
+async def test_async_ttl_zero_disables_caching(httpx_mock: HTTPXMock):
+ _mock_endpoints(httpx_mock)
+ async with SwgohComlinkAsync(url=BASE_URL, version_cache_ttl=0) as client:
+ await client.get_game_data()
+ await client.get_game_data()
+
+ assert len(_metadata_requests(httpx_mock)) == 2
+
+
+@pytest.mark.asyncio
+async def test_async_invalidate_and_refresh(httpx_mock: HTTPXMock, clock: dict[str, float]):
+ _mock_endpoints(httpx_mock)
+ async with SwgohComlinkAsync(url=BASE_URL, version_cache_ttl=60.0) as client:
+ await client.get_game_data()
+ client.invalidate_version_cache()
+ await client.get_game_data()
+ assert len(_metadata_requests(httpx_mock)) == 2
+
+ clock["t"] += 61.0
+ versions = await client.get_latest_game_data_version()
+ assert versions == {"game": "game-v1", "language": "lang-v1"}
+ assert len(_metadata_requests(httpx_mock)) == 3