Coverage for src/secchi/update.py: 86%

1"""Best-effort notification when a newer Secchi release is available.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7from collections.abc import Callable 

8from dataclasses import dataclass 

9from datetime import UTC, datetime, timedelta 

10from pathlib import Path 

11from typing import Any 

12 

13import httpx 

14from packaging.version import InvalidVersion, Version 

15 

16from secchi import __version__ 

17from secchi.cache import cache_root 

18 

19PYPI_URL = "https://pypi.org/pypi/secchi/json" 

20UPDATE_CACHE_TTL = timedelta(days=1) 

21UPDATE_CACHE_PATH = cache_root() / "update-check.json" 

22DISABLE_UPDATE_ENV = "SECCHI_DISABLE_UPDATE_CHECK" 

23 

24 

25@dataclass(frozen=True) 

26class UpdateNotice: 

27 """A newer release and the version currently running.""" 

28 

29 current_version: str 

30 latest_version: str 

31 

32 @property 

33 def message(self) -> str: 

34 return ( 

35 f"A newer Secchi version is available: {self.latest_version} " 

36 f"(current: {self.current_version}). Upgrade with: " 

37 "pipx upgrade secchi or uv tool upgrade secchi" 

38 ) 

39 

40 

41def update_check_disabled() -> bool: 

42 """Return whether the user opted out of release checks.""" 

43 

44 return os.environ.get(DISABLE_UPDATE_ENV, "").lower() in { 

45 "1", 

46 "true", 

47 "yes", 

48 "on", 

49 } 

50 

51 

52def check_for_update( 

53 *, 

54 current_version: str = __version__, 

55 cache_path: Path | None = None, 

56 now: Callable[[], datetime] | None = None, 

57 fetch: Callable[[], dict[str, Any]] | None = None, 

58) -> UpdateNotice | None: 

59 """Return a notice for a newer PyPI release, without failing the command. 

60 

61 The result is cached for one day. Network, filesystem, and malformed 

62 response errors intentionally produce no notice and never affect Secchi's 

63 primary command. 

64 """ 

65 

66 if update_check_disabled(): 

67 return None 

68 

69 clock = now or (lambda: datetime.now(UTC)) 

70 path = cache_path or UPDATE_CACHE_PATH 

71 payload = _load_cache(path, clock()) 

72 if payload is None: 

73 try: 

74 payload = (fetch or _fetch_latest)() 

75 except (OSError, httpx.HTTPError, ValueError, TypeError, KeyError): 

76 return None 

77 _save_cache(path, payload, clock()) 

78 

79 latest = payload.get("latest_version", "") 

80 if not isinstance(latest, str) or not _is_newer(latest, current_version): 

81 return None 

82 return UpdateNotice(current_version=current_version, latest_version=latest) 

83 

84 

85def _fetch_latest() -> dict[str, Any]: 

86 response = httpx.get(PYPI_URL, timeout=2.0) 

87 response.raise_for_status() 

88 data = response.json() 

89 return {"latest_version": data["info"]["version"]} 

90 

91 

92def _load_cache(path: Path, now: datetime) -> dict[str, Any] | None: 

93 try: 

94 data = json.loads(path.read_text()) 

95 fetched_at = datetime.fromisoformat(data["fetched_at"]) 

96 if now - fetched_at > UPDATE_CACHE_TTL: 

97 return None 

98 return data 

99 except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError): 

100 return None 

101 

102 

103def _save_cache(path: Path, payload: dict[str, Any], fetched_at: datetime) -> None: 

104 try: 

105 path.parent.mkdir(parents=True, exist_ok=True) 

106 path.write_text( 

107 json.dumps( 

108 {**payload, "fetched_at": fetched_at.isoformat()}, 

109 sort_keys=True, 

110 ) 

111 ) 

112 except OSError: 

113 pass 

114 

115 

116def _is_newer(latest: str, current: str) -> bool: 

117 try: 

118 return Version(latest) > Version(current) 

119 except InvalidVersion: 

120 return False