Coverage for src/data_api/api/routers/ping.py: 100%

1"""Readiness based on existing snapshots, without upstream probes.""" 

2 

3from datetime import datetime, timezone 

4from typing import Literal 

5 

6from fastapi import APIRouter, Response 

7 

8from data_api.api.schemas.ping import DatasetError, DatasetHealth, PingResponse 

9from data_api.core.config import PREFETCH_ENDPOINTS 

10from data_api.data.nthudata import DatasetState, Freshness 

11from data_api.runtime import get_runtime 

12 

13router = APIRouter() 

14 

15 

16def _build_dataset_health(state: DatasetState | None, now: float) -> DatasetHealth: 

17 if state is None: 

18 return DatasetHealth() 

19 snapshot = state.snapshot 

20 error = state.last_error 

21 return DatasetHealth( 

22 usable=state.usable, 

23 freshness=state.freshness, 

24 check_due=state.last_checked_at is None or now >= state.next_check, 

25 version=snapshot.version if snapshot else None, 

26 loaded_at=snapshot.loaded_at if snapshot else None, 

27 published_at=snapshot.published_at if snapshot else None, 

28 last_checked_at=state.last_checked_at, 

29 last_refresh_attempt_at=state.last_refresh_attempt_at, 

30 last_refresh_success_at=state.last_refresh_success_at, 

31 last_error=( 

32 DatasetError(category=error.category, at=error.at, status_code=error.status_code) 

33 if error 

34 else None 

35 ), 

36 ) 

37 

38 

39def _readiness_status(ready: bool, healthy: bool) -> Literal["ok", "degraded", "unavailable"]: 

40 if not ready: 

41 return "unavailable" 

42 return "ok" if healthy else "degraded" 

43 

44 

45@router.get("/ping", include_in_schema=False) 

46async def ping(response: Response) -> PingResponse: 

47 nthudata = get_runtime().datasets 

48 now = nthudata.clock() 

49 endpoints = {f"/{endpoint}" for endpoint in PREFETCH_ENDPOINTS} | nthudata.states.keys() 

50 datasets = { 

51 endpoint: _build_dataset_health(nthudata.states.get(endpoint), now) 

52 for endpoint in sorted(endpoints) 

53 } 

54 

55 ready = all(dataset.usable for dataset in datasets.values()) 

56 healthy = all( 

57 dataset.freshness == Freshness.CURRENT and not dataset.check_due 

58 for dataset in datasets.values() 

59 ) 

60 response.status_code = 200 if ready else 503 

61 response.headers["Cache-Control"] = "no-store" 

62 return PingResponse( 

63 status=_readiness_status(ready, healthy), 

64 ready=ready, 

65 checked_at=datetime.now(timezone.utc), 

66 datasets=datasets, 

67 ) 

68 

69 

70@router.head("/ping", include_in_schema=False) 

71async def ping_head() -> Response: 

72 response = Response(media_type="application/json") 

73 await ping(response) 

74 # An empty HEAD body must not advertise a zero-length GET representation. 

75 del response.headers["Content-Length"] 

76 return response