Coverage for src/ph/store_cache.py: 100.0%
88 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Structured, atomic game catalogue caches shared by every store."""
3import json
4import logging
5import os
6from collections.abc import Callable, Sequence
7from dataclasses import asdict, dataclass
8from pathlib import Path
9from time import time
10from typing import cast
11from urllib.parse import quote
13from ph.cache_policy import DEFAULT_CATALOGUE_TTL_DAYS, catalogue_ttl_seconds
14from ph.models import SearchResult
16LOGGER = logging.getLogger(__name__)
17CACHE_SCHEMA_VERSION = 1
19type CatalogueFetcher = Callable[[], Sequence[SearchResult]]
22class CatalogueCacheError(RuntimeError):
23 """A store catalogue could not be saved for offline use."""
26@dataclass(frozen=True, slots=True)
27class StoreCacheStatus:
28 """User-facing state of one store/platform catalogue file."""
30 path: Path
31 fetched_at: float
32 result_count: int
33 stale: bool
36@dataclass(frozen=True, slots=True)
37class _CachedCatalogue:
38 fetched_at: float
39 results: tuple[SearchResult, ...]
42class GameCatalogueCache:
43 """Cache complete store catalogues by stable store and platform identifiers."""
45 def __init__(
46 self,
47 root: Path,
48 store_id: str,
49 source_url: str,
50 ttl_seconds: int = catalogue_ttl_seconds(DEFAULT_CATALOGUE_TTL_DAYS),
51 ) -> None:
52 self.directory = root / "game-catalogues" / store_id
53 self.store_id = store_id
54 self.source_url = source_url.rstrip("/")
55 self.ttl_seconds = ttl_seconds
57 def _is_stale(self, cached: _CachedCatalogue) -> bool:
58 return time() - cached.fetched_at > self.ttl_seconds
60 def path_for(self, system_code: str) -> Path:
61 """Return a readable, filesystem-safe path for one platform catalogue."""
63 identifier = quote(system_code, safe="") if system_code else "all"
64 return self.directory / f"{identifier}.json"
66 def status(self, system_code: str) -> StoreCacheStatus | None:
67 """Describe a valid cache without contacting the store."""
69 path = self.path_for(system_code)
70 cached = self._read(path, system_code)
71 if cached is None:
72 return None
73 return StoreCacheStatus(
74 path, cached.fetched_at, len(cached.results), self._is_stale(cached)
75 )
77 def cached_files(self) -> tuple[Path, ...]:
78 """List structured catalogue files already stored for this source."""
80 try:
81 return tuple(sorted(self.directory.glob("*.json")))
82 except OSError:
83 return ()
85 def get_or_fetch(
86 self,
87 system_code: str,
88 fetcher: CatalogueFetcher,
89 *,
90 force: bool = False,
91 ) -> tuple[SearchResult, ...]:
92 """Use a fresh cache, refreshing expired data and retaining stale fallback data."""
94 path = self.path_for(system_code)
95 cached = self._read(path, system_code)
96 if cached is not None and not self._is_stale(cached) and not force:
97 LOGGER.debug(
98 "Using game catalogue cache store=%s system=%r results=%d",
99 self.store_id,
100 system_code,
101 len(cached.results),
102 )
103 return cached.results
104 try:
105 results = tuple(fetcher())
106 except OSError, TimeoutError, ValueError, RuntimeError:
107 if cached is None or force:
108 raise
109 LOGGER.warning(
110 "Store refresh failed; using stale catalogue store=%s system=%r",
111 self.store_id,
112 system_code,
113 exc_info=True,
114 )
115 return cached.results
116 self._write(path, system_code, results, strict=force)
117 return results
119 def _read(self, path: Path, system_code: str) -> _CachedCatalogue | None:
120 try:
121 payload = json.loads(path.read_text(encoding="utf-8"))
122 if (
123 not isinstance(payload, dict)
124 or payload.get("schema_version") != CACHE_SCHEMA_VERSION
125 or payload.get("store_id") != self.store_id
126 or payload.get("system_code") != system_code
127 or payload.get("source_url") != self.source_url
128 or not isinstance(payload.get("results"), list)
129 ):
130 return None
131 fetched_at = float(payload["fetched_at"])
132 results = tuple(_parse_result(item) for item in payload["results"])
133 except KeyError, OSError, TypeError, ValueError, json.JSONDecodeError:
134 return None
135 return _CachedCatalogue(fetched_at, results)
137 def _write(
138 self,
139 path: Path,
140 system_code: str,
141 results: tuple[SearchResult, ...],
142 *,
143 strict: bool,
144 ) -> None:
145 payload = {
146 "schema_version": CACHE_SCHEMA_VERSION,
147 "fetched_at": time(),
148 "store_id": self.store_id,
149 "system_code": system_code,
150 "source_url": self.source_url,
151 "results": [asdict(result) for result in results],
152 }
153 temporary = path.with_name(path.name + ".tmp")
154 try:
155 path.parent.mkdir(parents=True, exist_ok=True)
156 temporary.write_text(
157 json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
158 encoding="utf-8",
159 )
160 os.replace(temporary, path)
161 except OSError as error:
162 temporary.unlink(missing_ok=True)
163 if strict:
164 raise CatalogueCacheError(f"Could not save the game catalogue: {error}") from error
165 LOGGER.warning("Could not cache game catalogue path=%s: %s", path, error)
168def _parse_result(payload: object) -> SearchResult:
169 if not isinstance(payload, dict):
170 raise ValueError("invalid cached search result")
171 values = cast(dict[object, object], payload)
172 return SearchResult(
173 title=_required_string(values, "title"),
174 link=_required_string(values, "link"),
175 system=_required_string(values, "system"),
176 region=_required_string(values, "region"),
177 version=_required_string(values, "version"),
178 languages=_required_string(values, "languages"),
179 rating=_required_string(values, "rating"),
180 )
183def _required_string(payload: dict[object, object], field: str) -> str:
184 value = payload.get(field)
185 if not isinstance(value, str):
186 raise ValueError("invalid cached search result")
187 return value