Coverage for src/data_api/data/nthudata.py: 98%
268 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Disposable, instance-local snapshots of the authoritative published datasets."""
3import asyncio
4import hashlib
5import json
6import logging
7import os
8import socket
9import time
10from collections.abc import Callable
11from contextlib import asynccontextmanager
12from dataclasses import dataclass, field
13from datetime import datetime, timezone
14from enum import StrEnum
15from typing import Any, Generic, TypeVar
17import httpx
18from pydantic import BaseModel, ConfigDict, Field, ValidationError
20from data_api.core.exceptions import DataNotAvailableException
22logger = logging.getLogger(__name__)
23JsonData = dict | list
24T = TypeVar("T")
27class Freshness(StrEnum):
28 CURRENT = "current"
29 STALE = "stale"
30 UNVERIFIED = "unverified"
31 UNAVAILABLE = "unavailable"
34class FetchFailure(Exception):
35 """An expected upstream failure; never stores response bodies or credentials."""
37 def __init__(self, category: str, status_code: int | None = None):
38 super().__init__(category)
39 self.category = category
40 self.status_code = status_code
43@dataclass(frozen=True)
44class RefreshError:
45 category: str
46 at: datetime
47 status_code: int | None = None
50@dataclass(frozen=True)
51class Snapshot(Generic[T]):
52 raw: JsonData
53 data: T
54 version: str | None
55 loaded_at: datetime
56 published_at: str | None = None
59@dataclass
60class DatasetState(Generic[T]):
61 prepare: Callable[[JsonData], T]
62 install_prepared_json: bool = False
63 snapshot: Snapshot[T] | None = None
64 last_checked_at: datetime | None = None
65 last_refresh_attempt_at: datetime | None = None
66 last_refresh_success_at: datetime | None = None
67 last_error: RefreshError | None = None
68 freshness: Freshness = Freshness.UNAVAILABLE
69 next_check: float = 0
70 lock: asyncio.Lock = field(default_factory=asyncio.Lock)
72 @property
73 def usable(self) -> bool:
74 return self.snapshot is not None
77class DataFetcher:
78 """Reuse the client owned by the application lifespan."""
80 def __init__(self, base_url: str):
81 self.base_url = base_url
82 self.client: httpx.AsyncClient | None = None
84 async def fetch_json(self, url: str, sha256: str | None = None) -> JsonData:
85 if self.client is None:
86 raise FetchFailure("client_unavailable")
87 try:
88 response = await self.client.get(url)
89 response.raise_for_status()
90 except httpx.TimeoutException as exc:
91 raise FetchFailure("timeout") from exc
92 except httpx.HTTPStatusError as exc:
93 raise FetchFailure("http_status", exc.response.status_code) from exc
94 except httpx.ConnectError as exc:
95 cause: BaseException | None = exc
96 while cause is not None:
97 if isinstance(cause, socket.gaierror):
98 raise FetchFailure("dns") from exc
99 cause = cause.__cause__ or cause.__context__
100 raise FetchFailure("connection") from exc
101 except httpx.RequestError as exc:
102 raise FetchFailure("request") from exc
104 if sha256 is not None and hashlib.sha256(response.content).hexdigest() != sha256:
105 raise FetchFailure("checksum")
106 try:
107 data = response.json()
108 except (json.JSONDecodeError, UnicodeDecodeError) as exc:
109 raise FetchFailure("invalid_json") from exc
110 if not isinstance(data, (dict, list)):
111 raise FetchFailure("payload_type")
112 return data
115class ManifestEntry(BaseModel):
116 model_config = ConfigDict(extra="ignore", strict=True)
118 name: str = Field(min_length=1)
119 version: str | None = Field(default=None, min_length=1)
120 sha256: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
121 last_updated: str | None = None
123 @property
124 def active_version(self) -> str | None:
125 return self.version
128class PublishedManifest(BaseModel):
129 model_config = ConfigDict(extra="ignore", strict=True)
131 file_details: dict[str, list[ManifestEntry]]
134def parse_manifest(raw: JsonData) -> dict[str, ManifestEntry]:
135 try:
136 manifest = PublishedManifest.model_validate(raw)
137 except ValidationError as exc:
138 raise FetchFailure("manifest_parsing") from exc
139 entries = {}
140 for section, files in manifest.file_details.items():
141 for entry in files:
142 path = "/" + "/".join(part for part in (section.strip("/"), entry.name) if part)
143 if path in entries:
144 raise FetchFailure("manifest_parsing")
145 entries[path] = entry
146 return entries
149class FileDetailsManager:
150 """Cache successful checks AND failures, without presenting old metadata as fresh."""
152 def __init__(
153 self,
154 fetcher: DataFetcher,
155 file_details_url: str,
156 cache_expiry: float = 300,
157 clock: Callable[[], float] = time.monotonic,
158 ):
159 self.fetcher = fetcher
160 self.file_details_url = file_details_url
161 self.cache_expiry = cache_expiry
162 self.clock = clock
163 self.entries: dict[str, ManifestEntry] = {}
164 self.error: RefreshError | None = None
165 self.checked_at: datetime | None = None
166 self.next_check = 0.0
167 self.lock = asyncio.Lock()
169 async def check(self) -> None:
170 if self.checked_at is not None and self.clock() < self.next_check:
171 return
172 async with self.lock:
173 if self.checked_at is not None and self.clock() < self.next_check:
174 return
175 try:
176 raw = await self.fetcher.fetch_json(self.file_details_url)
177 entries = parse_manifest(raw)
178 except FetchFailure as exc:
179 category = (
180 "manifest_parsing"
181 if exc.category in {"invalid_json", "payload_type"}
182 else exc.category
183 )
184 self.error = RefreshError(category, datetime.now(timezone.utc), exc.status_code)
185 self.entries = {}
186 logger.warning(
187 "manifest check failed category=%s status=%s", category, exc.status_code
188 )
189 else:
190 self.entries = entries
191 self.error = None
192 logger.debug("manifest checked")
193 self.checked_at = datetime.now(timezone.utc)
194 self.next_check = self.clock() + self.cache_expiry
197class NTHUDataManager:
198 """One manager per running process; no disk cache or cross-instance coordination."""
200 def __init__(
201 self,
202 base_url: str | None = None,
203 file_details_cache_expiry: float = 300,
204 http_timeout: float = 15,
205 clock: Callable[[], float] = time.monotonic,
206 ):
207 if file_details_cache_expiry <= 0 or http_timeout <= 0:
208 raise ValueError("TTL and HTTP timeout must be positive")
209 self.base_url = (base_url or os.getenv("NTHU_DATA_URL", "https://data.nthusa.tw")).rstrip(
210 "/"
211 )
212 self.ttl = file_details_cache_expiry
213 self.http_timeout = http_timeout
214 self.clock = clock
215 self.fetcher = DataFetcher(self.base_url)
216 self.file_details_manager = FileDetailsManager(
217 self.fetcher, f"{self.base_url}/file_details.json", self.ttl, clock
218 )
219 self.states: dict[str, DatasetState[Any]] = {}
221 def register(self, endpoint: str, prepare: Callable[[JsonData], T]) -> DatasetState[T]:
222 key = self._normalize_endpoint_name(endpoint)
223 if key in self.states:
224 raise ValueError(f"Dataset already registered: {key}")
225 state = DatasetState(prepare)
226 self.states[key] = state
227 return state
229 @asynccontextmanager
230 async def lifespan(self, client: httpx.AsyncClient | None = None):
231 """Own and close the shared client, including on failed startup."""
232 if self.fetcher.client is not None:
233 raise RuntimeError("Data manager lifespan already running")
234 self.reset()
235 async with (
236 client
237 or httpx.AsyncClient(
238 http2=True, timeout=httpx.Timeout(self.http_timeout), follow_redirects=True
239 ) as active_client
240 ):
241 self.fetcher.client = active_client
242 try:
243 yield self
244 finally:
245 self.fetcher.client = None
246 self.reset()
248 def reset(self) -> None:
249 """A new instance has no last-known-good from its predecessor."""
250 for state in self.states.values():
251 state.snapshot = None
252 state.last_checked_at = None
253 state.last_refresh_attempt_at = None
254 state.last_refresh_success_at = None
255 state.last_error = None
256 state.freshness = Freshness.UNAVAILABLE
257 state.next_check = 0
258 state.lock = asyncio.Lock()
259 self.file_details_manager = FileDetailsManager(
260 self.fetcher, f"{self.base_url}/file_details.json", self.ttl, self.clock
261 )
263 def state_for(self, endpoint: str) -> DatasetState:
264 key = self._normalize_endpoint_name(endpoint)
265 if key not in self.states:
266 from data_api.data.validation import validate_dataset
268 state = self.register(key, lambda raw: validate_dataset(key, raw))
269 state.install_prepared_json = True
270 return self.states[key]
272 async def get_snapshot(self, endpoint: str, state: DatasetState[T]) -> Snapshot[T]:
273 if state.lock.locked() or state.last_checked_at is None or self.clock() >= state.next_check:
274 async with state.lock:
275 if state.last_checked_at is None or self.clock() >= state.next_check:
276 completed = False
277 try:
278 await self._refresh(self._normalize_endpoint_name(endpoint), state)
279 completed = True
280 finally:
281 if not completed:
282 state.next_check = 0
283 if state.snapshot is None:
284 raise DataNotAvailableException(
285 "Dataset temporarily unavailable. Please try again later."
286 )
287 return state.snapshot
289 def _record_load_failure(
290 self, endpoint: str, state: DatasetState[T], exc: FetchFailure
291 ) -> None:
292 state.last_error = RefreshError(exc.category, datetime.now(timezone.utc), exc.status_code)
293 state.freshness = Freshness.STALE if state.snapshot is not None else Freshness.UNAVAILABLE
294 logger.warning(
295 "dataset load failed dataset=%s category=%s status=%s usable=%s",
296 endpoint,
297 exc.category,
298 exc.status_code,
299 state.usable,
300 )
301 # Bound retry storms even when the first load has no fallback.
302 state.next_check = max(state.next_check, self.clock() + self.ttl)
304 @staticmethod
305 def _install_snapshot(
306 endpoint: str,
307 state: DatasetState[T],
308 raw: JsonData,
309 candidate: T,
310 version: str | None,
311 entry: ManifestEntry | None,
312 error: RefreshError | None,
313 ) -> None:
314 now = datetime.now(timezone.utc)
315 state.snapshot = Snapshot(
316 raw, candidate, version, now, entry.last_updated if entry else None
317 )
318 state.last_refresh_success_at = now
319 state.last_error = error
320 state.freshness = Freshness.UNVERIFIED if error else Freshness.CURRENT
321 logger.info(
322 "dataset load succeeded dataset=%s version=%s freshness=%s",
323 endpoint,
324 version,
325 state.freshness,
326 )
328 async def _refresh(self, endpoint: str, state: DatasetState[T]) -> None:
329 previous = state.snapshot
330 logger.debug("dataset check due dataset=%s", endpoint)
331 manifest = self.file_details_manager
332 await manifest.check()
333 state.last_checked_at = manifest.checked_at
334 entry = manifest.entries.get(endpoint)
335 version = entry.active_version if entry else None
336 error = manifest.error
337 if error is None and version is None:
338 error = RefreshError("version_unknown", datetime.now(timezone.utc))
339 # Anchor to the actual manifest check, not this reader's access time.
340 state.next_check = manifest.next_check
341 if error is not None and previous is not None:
342 state.last_error = error
343 state.freshness = Freshness.UNVERIFIED
344 logger.warning(
345 "dataset unverified, keeping snapshot dataset=%s category=%s",
346 endpoint,
347 error.category,
348 )
349 return
350 if version is not None and previous is not None and previous.version == version:
351 state.last_error = None
352 state.freshness = Freshness.CURRENT
353 logger.debug("dataset version unchanged dataset=%s version=%s", endpoint, version)
354 return
356 state.last_refresh_attempt_at = datetime.now(timezone.utc)
357 logger.info(
358 "dataset load started dataset=%s previous=%s new=%s",
359 endpoint,
360 previous.version if previous else None,
361 version,
362 )
363 try:
364 raw, candidate = await self._load_candidate(endpoint, state, entry)
365 except FetchFailure as exc:
366 self._record_load_failure(endpoint, state, exc)
367 else:
368 self._install_snapshot(endpoint, state, raw, candidate, version, entry, error)
370 async def _load_candidate(
371 self, endpoint: str, state: DatasetState[T], entry: ManifestEntry | None
372 ) -> tuple[JsonData, T]:
373 raw = await self.fetcher.fetch_json(
374 f"{self.base_url}{endpoint}", entry.sha256 if entry else None
375 )
376 try:
377 candidate = state.prepare(raw)
378 if state.install_prepared_json:
379 if not isinstance(candidate, (dict, list)):
380 raise FetchFailure("payload_type")
381 raw = candidate
382 return raw, candidate
383 except ValidationError as exc:
384 raise FetchFailure("validation") from exc
385 except ValueError as exc:
386 raise FetchFailure("transformation") from exc
388 async def get(self, endpoint_name: str) -> tuple[str | None, JsonData]:
389 snapshot = await self.get_snapshot(endpoint_name, self.state_for(endpoint_name))
390 return snapshot.version, snapshot.raw
392 async def prefetch(self, endpoints: list[str]) -> dict[str, bool]:
393 async def load(endpoint: str) -> bool:
394 try:
395 await self.get(endpoint)
396 except DataNotAvailableException:
397 return False
398 return True
400 tasks = {}
401 async with asyncio.TaskGroup() as group:
402 for endpoint in endpoints:
403 tasks[endpoint] = group.create_task(load(endpoint))
404 return {endpoint: task.result() for endpoint, task in tasks.items()}
406 @staticmethod
407 def _normalize_endpoint_name(endpoint_name: str) -> str:
408 return "/" + endpoint_name.lstrip("/")