Coverage for src/usls/adapters/outbound/espn_adapter.py: 100%

1"""Outbound adapter — translates domain calls into ESPN API HTTP requests. 

2 

3This is the only place in the codebase that knows about: 

4- The ESPN API host and URL structure 

5- How to issue HTTP requests and translate non-2xx responses into domain errors 

6 

7The wire-format → domain-model mapping lives in parsers.py so this module 

8stays focused on transport. 

9""" 

10 

11import asyncio 

12import logging 

13from typing import Any 

14 

15import httpx 

16 

17from ...domain.exceptions import UpstreamAPIError, USLSNotFoundError 

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

19from .parsers import ( 

20 _parse_article, 

21 _parse_match, 

22 _parse_match_details, 

23 _parse_player, 

24 _parse_standing, 

25 _parse_team, 

26) 

27 

28logger = logging.getLogger(__name__) 

29 

30_DEFAULT_BASE_URL = "https://site.api.espn.com" 

31_LEAGUE_PATH = "/apis/site/v2/sports/soccer/usa.w.usl.1" 

32# Standings live on the /apis/v2 surface — the /apis/site/v2 path returns an empty {}. 

33_STANDINGS_PATH = "/apis/v2/sports/soccer/usa.w.usl.1/standings" 

34 

35 

36def _check_response(response: httpx.Response, path: str) -> None: 

37 """Raise a domain exception for any non-2xx HTTP status. 

38 

39 Args: 

40 response: The httpx response to inspect. 

41 path: The request path, included in exception messages for context. 

42 

43 Raises: 

44 USLSNotFoundError: If the server returned HTTP 404. 

45 UpstreamAPIError: If the server returned any other 4xx or 5xx status. 

46 """ 

47 try: 

48 response.raise_for_status() 

49 except httpx.HTTPStatusError as exc: 

50 if exc.response.status_code == 404: 

51 raise USLSNotFoundError(f"Not found: {path}") from exc 

52 raise UpstreamAPIError(f"Upstream error {exc.response.status_code}: {path}") from exc 

53 

54 

55class ESPNAdapter: 

56 """Calls the ESPN public API for USL Super League data. 

57 

58 The underlying httpx.AsyncClient is created once at construction and reused 

59 for all requests so the TCP connection pool is retained across calls — 

60 avoiding a fresh TCP+TLS handshake on every API call. 

61 """ 

62 

63 def __init__(self, base_url: str = _DEFAULT_BASE_URL, client: httpx.AsyncClient | None = None) -> None: 

64 """Initialize the adapter with an optional HTTP client. 

65 

66 Args: 

67 base_url: Base URL of the ESPN API. Defaults to https://site.api.espn.com. 

68 client: An httpx.AsyncClient instance to reuse across all requests. 

69 Inject a pre-configured mock in tests. 

70 """ 

71 self._client = client or httpx.AsyncClient(base_url=base_url, timeout=30.0) 

72 

73 async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: 

74 """Execute a GET request and return the parsed JSON body. 

75 

76 Raises: 

77 USLSNotFoundError: If the server returns HTTP 404. 

78 UpstreamAPIError: If the server returns any other 4xx or 5xx response. 

79 """ 

80 logger.debug("GET %s params=%s", path, params) 

81 response = await self._client.get(path, params=params or {}) 

82 _check_response(response, path) 

83 return response.json() 

84 

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

86 """Return all active USL Super League teams.""" 

87 data = await self._get(f"{_LEAGUE_PATH}/teams", {"limit": 100}) 

88 raw_teams = data.get("sports", [{}])[0].get("leagues", [{}])[0].get("teams", []) 

89 return [_parse_team(t) for t in raw_teams] 

90 

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

92 """Return a single team by its ESPN team ID. 

93 

94 Raises: 

95 USLSNotFoundError: If no team with that ID exists. 

96 """ 

97 data = await self._get(f"{_LEAGUE_PATH}/teams/{team_id}") 

98 raw = data.get("team") 

99 if not raw: 

100 raise USLSNotFoundError(f"Team not found: {team_id}") 

101 return _parse_team(raw) 

102 

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

104 """Return matches on the given date or date range, or the current week if date is None.""" 

105 params: dict[str, Any] = {} 

106 if date and end_date: 

107 params["dates"] = f"{date}-{end_date}" 

108 elif date: 

109 params["dates"] = date 

110 data = await self._get(f"{_LEAGUE_PATH}/scoreboard", params) 

111 return [_parse_match(e) for e in data.get("events", [])] 

112 

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

114 """Return the active roster for a team. 

115 

116 Raises: 

117 USLSNotFoundError: If no team with that ID exists. 

118 """ 

119 data = await self._get(f"{_LEAGUE_PATH}/teams/{team_id}/roster") 

120 return [_parse_player(p) for p in data.get("athletes", [])] 

121 

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

123 """Return detailed information for a single match. 

124 

125 Raises: 

126 USLSNotFoundError: If no match with that ID exists. 

127 """ 

128 data = await self._get(f"{_LEAGUE_PATH}/summary", {"event": match_id}) 

129 return _parse_match_details(data) 

130 

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

132 """Return all scheduled, in-progress, and completed matches for a team. 

133 

134 ESPN's team-schedule endpoint returns only past events by default and only 

135 upcoming events when called with ``fixture=true``. Both variants are fetched 

136 in parallel and merged, deduped by event id, and sorted chronologically so 

137 callers see the full season in calendar order. 

138 

139 Raises: 

140 USLSNotFoundError: If no team with that ID exists. 

141 """ 

142 path = f"{_LEAGUE_PATH}/teams/{team_id}/schedule" 

143 past, future = await asyncio.gather( 

144 self._get(path), 

145 self._get(path, {"fixture": "true"}), 

146 ) 

147 events_by_id: dict[str, dict[str, Any]] = {} 

148 for event in (*past.get("events", []), *future.get("events", [])): 

149 event_id = str(event.get("id", "")) 

150 events_by_id.setdefault(event_id, event) 

151 ordered = sorted(events_by_id.values(), key=lambda e: e.get("date", "")) 

152 return [_parse_match(e) for e in ordered] 

153 

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

155 """Return recent USL Super League news articles.""" 

156 data = await self._get(f"{_LEAGUE_PATH}/news", {"limit": limit}) 

157 return [_parse_article(a) for a in data.get("articles", [])] 

158 

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

160 """Return the current USL Super League standings ordered by points descending.""" 

161 data = await self._get(_STANDINGS_PATH) 

162 entries: list[dict[str, Any]] = [] 

163 for season in data.get("children", []): 

164 for division in season.get("standings", {}).get("entries", []): 

165 entries.append(division) 

166 standings = [s for e in entries if (s := _parse_standing(e)) is not None] 

167 return sorted(standings, key=lambda s: s.points, reverse=True)