Coverage for src/fastapi_cloud_cli/utils/version_check.py: 100%

109 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-11-09 00:25 +0000

1import json 1abcdefg

2import logging 1abcdefg

3import re 1abcdefg

4import threading 1abcdefg

5from contextlib import suppress 1abcdefg

6from dataclasses import dataclass 1abcdefg

7from datetime import datetime, timedelta, timezone 1abcdefg

8from pathlib import Path 1abcdefg

9 

10import httpx 1abcdefg

11from detect_installer import detect_installer 1abcdefg

12from pydantic import AwareDatetime, BaseModel, ValidationError 1abcdefg

13 

14from fastapi_cloud_cli import __version__ 1abcdefg

15from fastapi_cloud_cli.utils.config import get_version_check_cache_path 1abcdefg

16 

17logger = logging.getLogger(__name__) 1abcdefg

18 

19PACKAGE_NAME = "fastapi-cloud-cli" 1abcdefg

20DEFAULT_UPGRADE_COMMAND = f"pip install --upgrade {PACKAGE_NAME}" 1abcdefg

21PYPI_JSON_URL = f"https://pypi.org/pypi/{PACKAGE_NAME}/json" 1abcdefg

22VERSION_CHECK_TIMEOUT_SECONDS = 2.0 1abcdefg

23VERSION_CHECK_JOIN_TIMEOUT_SECONDS = 0.2 1abcdefg

24VERSION_CHECK_CACHE_TTL = timedelta(hours=24) 1abcdefg

25DISABLE_VERSION_CHECK_ENV = "FASTAPI_CLOUD_DISABLE_VERSION_CHECK" 1abcdefg

26SIMPLE_RELEASE_VERSION_RE = re.compile(r"\d+(?:\.\d+)*") 1abcdefg

27 

28 

29@dataclass(frozen=True) 1abcdefg

30class VersionUpdate: 1abcdefg

31 current: str 1abcdefg

32 latest: str 1abcdefg

33 

34 

35class VersionCheckCache(BaseModel): 1abcdefg

36 latest_version: str 1abcdefg

37 checked_at: AwareDatetime 1abcdefg

38 

39 

40class PyPIProjectInfo(BaseModel): 1abcdefg

41 version: str 1abcdefg

42 

43 

44class PyPIProjectResponse(BaseModel): 1abcdefg

45 info: PyPIProjectInfo 1abcdefg

46 

47 

48def _parse_simple_release_version(version: str) -> tuple[int, ...] | None: 1abcdefg

49 if not SIMPLE_RELEASE_VERSION_RE.fullmatch(version): 2h i C D E # j cbjbkbk l F G H $ m dblbmbn o I J K % p ebnbobq r L M N ' s fbpbqbt u O P Q ( v gbrbsbw x R S T ) y hbtbubz A U V W * B ibvbwb

50 logger.debug("Skipping non-simple version string: %r", version) 2cbjbdblbebnbfbpbgbrbhbtbibvb

51 return None 2cbjbdblbebnbfbpbgbrbhbtbibvb

52 

53 return tuple(int(part) for part in version.split(".")) 2h i C D E # j cbkbk l F G H $ m dbmbn o I J K % p ebobq r L M N ' s fbqbt u O P Q ( v gbsbw x R S T ) y hbubz A U V W * B ibwb

54 

55 

56def is_newer_version(latest: str, current: str) -> bool: 1abcdefg

57 latest_parts = _parse_simple_release_version(latest) 2h i C D E # j cbk l F G H $ m dbn o I J K % p ebq r L M N ' s fbt u O P Q ( v gbw x R S T ) y hbz A U V W * B ib

58 current_parts = _parse_simple_release_version(current) 2h i C D E # j cbk l F G H $ m dbn o I J K % p ebq r L M N ' s fbt u O P Q ( v gbw x R S T ) y hbz A U V W * B ib

59 

60 if latest_parts is None or current_parts is None: 2h i C D E # j cbk l F G H $ m dbn o I J K % p ebq r L M N ' s fbt u O P Q ( v gbw x R S T ) y hbz A U V W * B ib

61 return False 2cbdbebfbgbhbib

62 

63 return latest_parts > current_parts 2h i C D E # j cbk l F G H $ m dbn o I J K % p ebq r L M N ' s fbt u O P Q ( v gbw x R S T ) y hbz A U V W * B ib

64 

65 

66def read_cached_latest_version( 1abcdefg

67 cache_path: Path, 

68 *, 

69 ttl: timedelta = VERSION_CHECK_CACHE_TTL, 

70) -> str | None: 

71 now = datetime.now(timezone.utc) 2h i X 4 C D E # j xbjbkbk l Y 5 F G H $ m yblbmbn o Z 6 I J K % p zbnbobq r 0 7 L M N ' s Abpbqbt u 1 8 O P Q ( v Bbrbsbw x 2 9 R S T ) y Cbtbubz A 3 ! U V W * B Dbvbwb

72 

73 try: 2h i X 4 C D E # j xbjbkbk l Y 5 F G H $ m yblbmbn o Z 6 I J K % p zbnbobq r 0 7 L M N ' s Abpbqbt u 1 8 O P Q ( v Bbrbsbw x 2 9 R S T ) y Cbtbubz A 3 ! U V W * B Dbvbwb

74 cache = VersionCheckCache.model_validate_json( 2h i X 4 C D E # j xbjbkbk l Y 5 F G H $ m yblbmbn o Z 6 I J K % p zbnbobq r 0 7 L M N ' s Abpbqbt u 1 8 O P Q ( v Bbrbsbw x 2 9 R S T ) y Cbtbubz A 3 ! U V W * B Dbvbwb

75 cache_path.read_text(encoding="utf-8") 

76 ) 

77 except (OSError, ValidationError) as error: 2X 4 D E xbY 5 G H ybZ 6 J K zb0 7 M N Ab1 8 P Q Bb2 9 S T Cb3 ! V W Db

78 logger.debug("Could not read CLI version cache: %s", error) 2X 4 D E xbY 5 G H ybZ 6 J K zb0 7 M N Ab1 8 P Q Bb2 9 S T Cb3 ! V W Db

79 return None 2X 4 D E xbY 5 G H ybZ 6 J K zb0 7 M N Ab1 8 P Q Bb2 9 S T Cb3 ! V W Db

80 

81 if _parse_simple_release_version(cache.latest_version) is None: 2h i C # j jbkbk l F $ m lbmbn o I % p nbobq r L ' s pbqbt u O ( v rbsbw x R ) y tbubz A U * B vbwb

82 return None 2jblbnbpbrbtbvb

83 

84 if now - cache.checked_at > ttl: 2h i C # j kbk l F $ m mbn o I % p obq r L ' s qbt u O ( v sbw x R ) y ubz A U * B wb

85 return None 2C kbF mbI obL qbO sbR ubU wb

86 

87 return cache.latest_version 1hi#jkl$mno%pqr'stu(vwx)yzA*B

88 

89 

90def write_latest_version_cache( 1abcdefg

91 cache_path: Path, 

92 *, 

93 latest_version: str, 

94 now: datetime | None = None, 

95) -> None: 

96 now = now or datetime.now(timezone.utc) 2h i C D E j LbMbEbk l F G H m NbObFbn o I J K p PbQbGbq r L M N s RbSbHbt u O P Q v TbUbIbw x R S T y VbWbJbz A U V W B XbYbKb

97 data = { 2h i C D E j LbMbEbk l F G H m NbObFbn o I J K p PbQbGbq r L M N s RbSbHbt u O P Q v TbUbIbw x R S T y VbWbJbz A U V W B XbYbKb

98 "latest_version": latest_version, 

99 "checked_at": now.isoformat(), 

100 } 

101 

102 try: 2h i C D E j LbMbEbk l F G H m NbObFbn o I J K p PbQbGbq r L M N s RbSbHbt u O P Q v TbUbIbw x R S T y VbWbJbz A U V W B XbYbKb

103 cache_path.parent.mkdir(parents=True, exist_ok=True) 2h i C D E j LbMbEbk l F G H m NbObFbn o I J K p PbQbGbq r L M N s RbSbHbt u O P Q v TbUbIbw x R S T y VbWbJbz A U V W B XbYbKb

104 cache_path.write_text(json.dumps(data), encoding="utf-8") 2h i C D E j LbMbk l F G H m NbObn o I J K p PbQbq r L M N s RbSbt u O P Q v TbUbw x R S T y VbWbz A U V W B XbYb

105 except OSError as error: 2EbFbGbHbIbJbKb

106 logger.debug("Could not write CLI version cache: %s", error) 2EbFbGbHbIbJbKb

107 

108 

109def fetch_latest_version() -> str | None: 1abcdefg

110 headers = {"User-Agent": f"fastapi-cloud-cli/{__version__}"} 1X4CDEY5FGHZ6IJK07LMN18OPQ29RST3!UVW

111 

112 try: 1X4CDEY5FGHZ6IJK07LMN18OPQ29RST3!UVW

113 with httpx.Client( 1X4CDEY5FGHZ6IJK07LMN18OPQ29RST3!UVW

114 timeout=httpx.Timeout(VERSION_CHECK_TIMEOUT_SECONDS), 

115 headers=headers, 

116 ) as client: 

117 response = client.get(PYPI_JSON_URL) 1X4CDEY5FGHZ6IJK07LMN18OPQ29RST3!UVW

118 response.raise_for_status() 1XCDEYFGHZIJK0LMN1OPQ2RST3UVW

119 data = PyPIProjectResponse.model_validate_json(response.text) 1XCDEYFGHZIJK0LMN1OPQ2RST3UVW

120 except (httpx.HTTPError, ValidationError) as error: 1X4Y5Z60718293!

121 logger.debug("Could not check latest CLI version: %s", error) 1X4Y5Z60718293!

122 return None 1X4Y5Z60718293!

123 

124 return data.info.version 1CDEFGHIJKLMNOPQRSTUVW

125 

126 

127def check_for_update() -> VersionUpdate | None: 1abcdefg

128 cache_path = get_version_check_cache_path() 1hiX4CDE#jklY5FGH$mnoZ6IJK%pqr07LMN'stu18OPQ(vwx29RST)yzA3!UVW*B

129 

130 if (latest_version := read_cached_latest_version(cache_path)) is None: 1hiX4CDE#jklY5FGH$mnoZ6IJK%pqr07LMN'stu18OPQ(vwx29RST)yzA3!UVW*B

131 if (latest_version := fetch_latest_version()) is None: 1X4CDEY5FGHZ6IJK07LMN18OPQ29RST3!UVW

132 return None 1X4Y5Z60718293!

133 

134 write_latest_version_cache( 1CDEFGHIJKLMNOPQRSTUVW

135 cache_path, 

136 latest_version=latest_version, 

137 ) 

138 

139 if not is_newer_version(latest_version, __version__): 1hiCDE#jklFGH$mnoIJK%pqrLMN'stuOPQ(vwxRST)yzAUVW*B

140 return None 1DGJMPSV

141 

142 return VersionUpdate(current=__version__, latest=latest_version) 1hiCE#jklFH$mnoIK%pqrLN'stuOQ(vwxRT)yzAUW*B

143 

144 

145def get_upgrade_command() -> str: 1abcdefg

146 installer_info = detect_installer(PACKAGE_NAME) 2h i + j Zb%b0bk l , m 1b'b2bn o - p 3b(b4bq r . s 5b)b6bt u / v 7b*b8bw x : y 9b+b!bz A ; B #b,b$b

147 

148 if installer_info is None or installer_info.upgrade_cmd is None: 2h i + j Zb0bk l , m 1b2bn o - p 3b4bq r . s 5b6bt u / v 7b8bw x : y 9b!bz A ; B #b$b

149 return DEFAULT_UPGRADE_COMMAND 2Zb1b3b5b7b9b#b

150 

151 return installer_info.upgrade_cmd 2h i + j 0bk l , m 2bn o - p 4bq r . s 6bt u / v 8bw x : y !bz A ; B $b

152 

153 

154def format_update_message( 1abcdefg

155 update: VersionUpdate, 

156) -> str: 

157 return ( 2h i + -bj k l , .bm n o - /bp q r . :bs t u / ;bv w x : =by z A ; ?bB

158 "A newer FastAPI Cloud CLI version is available: " 

159 f"{update.current} → [bold]{update.latest}[/]\n\n" 

160 f'Run "[blue]{get_upgrade_command()}[/]" to upgrade.' 

161 ) 

162 

163 

164class BackgroundVersionCheck: 1abcdefg

165 def __init__(self) -> None: 1abcdefg

166 self._thread = threading.Thread(target=self._run, daemon=True) 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

167 self._update: VersionUpdate | None = None 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

168 self._message_returned = False 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

169 

170 def start(self) -> None: 1abcdefg

171 self._thread.start() 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

172 

173 def _run(self) -> None: 1abcdefg

174 with suppress(Exception): 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

175 self._update = check_for_update() 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

176 

177 def get_update_message(self) -> str | None: 1abcdefg

178 if self._message_returned: 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

179 return None 1+,-./:;

180 

181 self._thread.join(timeout=VERSION_CHECK_JOIN_TIMEOUT_SECONDS) 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

182 

183 if self._update: 2h i = ? + j k l @ [ , m n o ] ^ - p q r _ ` . s t u { | / v w x } ~ : y z A abbb; B

184 self._message_returned = True 1hi+jkl,mno-pqr.stu/vwx:yzA;B

185 return format_update_message(self._update) 1hi+jkl,mno-pqr.stu/vwx:yzA;B

186 

187 return None 2= ? @ [ ] ^ _ ` { | } ~ abbb