Coverage for src/qdrant_loader/utils/version_check.py: 95%

1"""Version checking utility for QDrant Loader CLI.""" 

2 

3import json 

4import time 

5from pathlib import Path 

6from urllib.error import HTTPError, URLError 

7from urllib.request import Request, urlopen 

8 

9from packaging import version 

10 

11 

12class VersionChecker: 

13 """Check for new versions of qdrant-loader on PyPI.""" 

14 

15 PYPI_API_URL = "https://pypi.org/pypi/qdrant-loader/json" 

16 CACHE_FILE = ".qdrant_loader_version_cache" 

17 CACHE_DURATION = 24 * 60 * 60 # 24 hours in seconds 

18 

19 def __init__(self, current_version: str): 

20 """Initialize version checker. 

21 

22 Args: 

23 current_version: Current version of qdrant-loader 

24 """ 

25 self.current_version = current_version 

26 self.cache_path = Path.home() / self.CACHE_FILE 

27 

28 def _get_cache_data(self) -> dict | None: 

29 """Get cached version data if still valid. 

30 

31 Returns: 

32 Cached data if valid, None otherwise 

33 """ 

34 try: 

35 if not self.cache_path.exists(): 

36 return None 

37 

38 with open(self.cache_path) as f: 

39 cache_data = json.load(f) 

40 

41 # Check if cache is still valid 

42 cache_time = cache_data.get("timestamp", 0) 

43 if time.time() - cache_time > self.CACHE_DURATION: 

44 return None 

45 

46 return cache_data 

47 except (json.JSONDecodeError, OSError): 

48 return None 

49 

50 def _save_cache_data(self, data: dict) -> None: 

51 """Save version data to cache. 

52 

53 Args: 

54 data: Data to cache 

55 """ 

56 try: 

57 cache_data = {"timestamp": time.time(), **data} 

58 with open(self.cache_path, "w") as f: 

59 json.dump(cache_data, f) 

60 except OSError: 

61 # Silently fail if we can't write cache 

62 pass 

63 

64 def _fetch_latest_version(self) -> str | None: 

65 """Fetch latest version from PyPI. 

66 

67 Returns: 

68 Latest version string or None if fetch failed 

69 """ 

70 try: 

71 # Create request with user agent 

72 req = Request( 

73 self.PYPI_API_URL, 

74 headers={"User-Agent": f"qdrant-loader/{self.current_version}"}, 

75 ) 

76 

77 # Set timeout to avoid hanging 

78 with urlopen(req, timeout=5) as response: 

79 data = json.loads(response.read().decode("utf-8")) 

80 return data["info"]["version"] 

81 except (URLError, HTTPError, json.JSONDecodeError, KeyError, OSError): 

82 return None 

83 

84 def check_for_updates(self, silent: bool = False) -> tuple[bool, str | None]: 

85 """Check if a newer version is available. 

86 

87 Args: 

88 silent: If True, don't show any output 

89 

90 Returns: 

91 Tuple of (has_update, latest_version) 

92 """ 

93 # Skip check if current version is unknown 

94 if self.current_version in ("unknown", "Unknown"): 

95 return False, None 

96 

97 # Try to get from cache first 

98 cache_data = self._get_cache_data() 

99 if cache_data: 

100 latest_version = cache_data.get("latest_version") 

101 else: 

102 # Fetch from PyPI 

103 latest_version = self._fetch_latest_version() 

104 if latest_version: 

105 self._save_cache_data({"latest_version": latest_version}) 

106 

107 if not latest_version: 

108 return False, None 

109 

110 try: 

111 # Compare versions 

112 current_ver = version.parse(self.current_version) 

113 latest_ver = version.parse(latest_version) 

114 

115 has_update = latest_ver > current_ver 

116 return has_update, latest_version 

117 except version.InvalidVersion: 

118 return False, None 

119 

120 def show_update_notification(self, latest_version: str) -> None: 

121 """Show update notification to user. 

122 

123 Args: 

124 latest_version: Latest available version 

125 """ 

126 print("\n🆕 A new version of qdrant-loader is available!") 

127 print(f" Current: {self.current_version}") 

128 print(f" Latest: {latest_version}") 

129 print( 

130 " Update: pip install --upgrade qdrant-loader qdrant-loader-mcp-server" 

131 ) 

132 print() 

133 

134 

135def check_version_async(current_version: str, silent: bool = False) -> None: 

136 """Asynchronously check for version updates. 

137 

138 Args: 

139 current_version: Current version of qdrant-loader 

140 silent: If True, don't show any output 

141 """ 

142 

143 def _check(): 

144 checker = VersionChecker(current_version) 

145 has_update, latest_version = checker.check_for_updates(silent=silent) 

146 

147 if has_update and latest_version and not silent: 

148 checker.show_update_notification(latest_version) 

149 

150 # Run check in background thread to avoid blocking CLI 

151 try: 

152 import threading 

153 

154 thread = threading.Thread(target=_check, daemon=True) 

155 thread.start() 

156 except Exception: 

157 # Silently fail if threading doesn't work 

158 pass