Coverage for src/usls/adapters/outbound/caching_adapter.py: 96%

1"""CachingAdapter — transparent TTL-cache decorator for USLSAPIPort. 

2 

3Wraps any USLSAPIPort implementation and caches successful results in a 

4plain dict. Cache keys are derived from the method name plus the sorted 

5keyword arguments, making the cache transparent to callers. 

6 

7The clock function is injectable so tests can control TTL expiry without 

8actually sleeping. 

9 

10USL Super League data changes infrequently (lineups, scores update during matches), so 

11a default TTL of 5 minutes is appropriate for most tools. The scoreboard 

12uses a shorter TTL of 60 seconds to stay reasonably live during matches. 

13""" 

14 

15import json 

16import logging 

17import time 

18from collections.abc import Callable 

19from typing import Any 

20 

21from ...domain.models import Match, MatchDetails, NewsArticle, Player, Standing, Team 

22from ...ports.outbound import USLSAPIPort 

23 

24logger = logging.getLogger(__name__) 

25 

26 

27def _cache_key(method: str, kwargs: dict[str, Any]) -> str: 

28 """Build a deterministic cache key from a method name and its kwargs. 

29 

30 Args: 

31 method: The name of the port method being called. 

32 kwargs: The keyword arguments passed to that method. 

33 

34 Returns: 

35 A JSON string that uniquely identifies this (method, params) pair. 

36 """ 

37 return json.dumps({"method": method, "params": kwargs}, sort_keys=True, default=str) 

38 

39 

40class CachingAdapter: 

41 """Decorates a USLSAPIPort with a TTL-based in-memory cache.""" 

42 

43 def __init__( 

44 self, 

45 inner: USLSAPIPort, 

46 ttl_seconds: float = 300.0, 

47 scoreboard_ttl_seconds: float = 60.0, 

48 now: Callable[[], float] = time.monotonic, 

49 ) -> None: 

50 """Initialize the caching adapter. 

51 

52 Args: 

53 inner: The USLSAPIPort implementation to wrap. 

54 ttl_seconds: Default time-to-live for cached entries in seconds. 

55 scoreboard_ttl_seconds: TTL for scoreboard data (shorter for live match updates). 

56 now: Callable that returns the current monotonic time. Injectable for testing. 

57 """ 

58 self._inner = inner 

59 self._ttl = ttl_seconds 

60 self._scoreboard_ttl = scoreboard_ttl_seconds 

61 self._now = now 

62 self._cache: dict[str, tuple[float, Any]] = {} 

63 

64 async def _get_or_fetch(self, method_name: str, ttl: float, **kwargs: Any) -> Any: 

65 """Return a cached result or fetch from the inner adapter. 

66 

67 Args: 

68 method_name: Name of the method on the inner adapter to call. 

69 ttl: TTL in seconds to use for this cache entry. 

70 **kwargs: Keyword arguments forwarded to the method. 

71 

72 Returns: 

73 The cached or freshly-fetched result. 

74 """ 

75 key = _cache_key(method_name, kwargs) 

76 entry = self._cache.get(key) 

77 if entry is not None: 

78 expiry, result = entry 

79 if self._now() < expiry: 

80 logger.debug("Cache hit for %s", method_name) 

81 return result 

82 del self._cache[key] 

83 

84 logger.debug("Cache miss for %s, fetching from inner adapter", method_name) 

85 method = getattr(self._inner, method_name) 

86 result = await method(**kwargs) 

87 self._cache[key] = (self._now() + ttl, result) 

88 return result 

89 

90 async def get_teams(self) -> list[Team]: 

91 return await self._get_or_fetch("get_teams", self._ttl) 

92 

93 async def get_team(self, team_id: str) -> Team: 

94 return await self._get_or_fetch("get_team", self._ttl, team_id=team_id) 

95 

96 async def get_scoreboard(self, date: str | None = None, end_date: str | None = None) -> list[Match]: 

97 return await self._get_or_fetch("get_scoreboard", self._scoreboard_ttl, date=date, end_date=end_date) 

98 

99 async def get_team_schedule(self, team_id: str) -> list[Match]: 

100 return await self._get_or_fetch("get_team_schedule", self._ttl, team_id=team_id) 

101 

102 async def get_match_details(self, match_id: str) -> MatchDetails: 

103 return await self._get_or_fetch("get_match_details", self._scoreboard_ttl, match_id=match_id) 

104 

105 async def get_roster(self, team_id: str) -> list[Player]: 

106 return await self._get_or_fetch("get_roster", self._ttl, team_id=team_id) 

107 

108 async def get_news(self, limit: int) -> list[NewsArticle]: 

109 return await self._get_or_fetch("get_news", self._ttl, limit=limit) 

110 

111 async def get_standings(self) -> list[Standing]: 

112 return await self._get_or_fetch("get_standings", self._ttl)