Coverage for src/epl/adapters/outbound/pulselive_adapter.py: 45%

1"""Pulselive adapter — talks to ``footballapi.pulselive.com`` for richer PL data. 

2 

3Pulselive is the upstream powering premierleague.com's front-end. Its 

4``/players`` endpoint returns Opta-cross-referenced player rows with data 

5ESPN's roster feed does not carry (shirt number, positional detail, loan 

6flag, DoB, Opta player id). Access requires an ``Origin`` header matching 

7``https://www.premierleague.com`` — no key, but the CORS gate is enforced. 

8""" 

9 

10import logging 

11from typing import Any 

12 

13import httpx 

14 

15from ...domain.exceptions import EPLNotFoundError, UpstreamAPIError 

16from ...domain.models import PulselivePlayer 

17from .pulselive_parsers import _parse_pulselive_player 

18 

19logger = logging.getLogger(__name__) 

20 

21_DEFAULT_BASE_URL = "https://footballapi.pulselive.com" 

22# The Premier League Pulselive competition id is stable at 1. 

23_PL_COMPETITION_ID = 1 

24_REQUIRED_HEADERS = { 

25 "Origin": "https://www.premierleague.com", 

26 "Referer": "https://www.premierleague.com/", 

27} 

28 

29 

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

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

32 try: 

33 response.raise_for_status() 

34 except httpx.HTTPStatusError as exc: 

35 if exc.response.status_code == 404: 

36 raise EPLNotFoundError(f"Not found: {path}") from exc 

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

38 

39 

40class PulseliveAdapter: 

41 """Calls the Pulselive/Opta-backed API for Premier League data. 

42 

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

44 for all requests so the TCP connection pool is retained across calls. 

45 """ 

46 

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

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

49 

50 Args: 

51 base_url: Base URL of the Pulselive API. Defaults to https://footballapi.pulselive.com. 

52 client: Optional pre-configured httpx.AsyncClient. Inject in tests. 

53 """ 

54 self._client = client or httpx.AsyncClient(base_url=base_url, timeout=30.0, headers=_REQUIRED_HEADERS) 

55 

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

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

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

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

60 _check_response(response, path) 

61 return response.json() 

62 

63 async def latest_completed_season_id(self) -> int: 

64 """Return the id of the most recently completed PL season. 

65 

66 Uses the compseasons endpoint's date fields — the first entry whose 

67 end date is in the past is treated as the most recent completed 

68 season. Falls back to the second entry (skipping the in-flight one) 

69 when no explicit date is available. 

70 """ 

71 data = await self._get(f"/football/competitions/{_PL_COMPETITION_ID}/compseasons") 

72 seasons = data.get("content", []) 

73 if not seasons: 

74 raise EPLNotFoundError("No Premier League seasons advertised") 

75 completed = [s for s in seasons if _has_completed(s)] 

76 target = completed[0] if completed else seasons[1] if len(seasons) > 1 else seasons[0] 

77 return int(target["id"]) 

78 

79 async def get_players(self, season_id: int, page_size: int) -> list[PulselivePlayer]: 

80 """Return up to ``page_size`` players for ``season_id``.""" 

81 data = await self._get( 

82 "/football/players", 

83 {"compSeasons": season_id, "pageSize": page_size, "altIds": "true"}, 

84 ) 

85 rows = data.get("content", []) 

86 return [_parse_pulselive_player(r) for r in rows] 

87 

88 

89def _has_completed(season: dict[str, Any]) -> bool: 

90 """Return True if the season carries an end date and no start date in the future. 

91 

92 Season objects may lack date fields entirely on legacy entries; those are 

93 treated as completed on the assumption that the compseasons list is 

94 ordered newest-first (the observed shape at ``/compseasons``). 

95 """ 

96 label = season.get("label") or "" 

97 # The current-season label matches "English Premier League Season YYYY/YYYY" 

98 # while completed labels use the shorter "YYYY/YY" form. 

99 return "Season" not in label