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

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

2 

3import asyncio 

4import json 

5import os 

6import time 

7from pathlib import Path 

8from typing import Optional, Tuple 

9from urllib.parse import urljoin 

10from urllib.request import Request, urlopen 

11from urllib.error import URLError, HTTPError 

12 

13from packaging import version 

14 

15 

16class VersionChecker: 

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

18 

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

20 CACHE_FILE = ".qdrant_loader_version_cache" 

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

22 

23 def __init__(self, current_version: str): 

24 """Initialize version checker. 

25 

26 Args: 

27 current_version: Current version of qdrant-loader 

28 """ 

29 self.current_version = current_version 

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

31 

32 def _get_cache_data(self) -> Optional[dict]: 

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

34 

35 Returns: 

36 Cached data if valid, None otherwise 

37 """ 

38 try: 

39 if not self.cache_path.exists(): 

40 return None 

41 

42 with open(self.cache_path, "r") as f: 

43 cache_data = json.load(f) 

44 

45 # Check if cache is still valid 

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

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

48 return None 

49 

50 return cache_data 

51 except (json.JSONDecodeError, OSError): 

52 return None 

53 

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

55 """Save version data to cache. 

56 

57 Args: 

58 data: Data to cache 

59 """ 

60 try: 

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

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

63 json.dump(cache_data, f) 

64 except OSError: 

65 # Silently fail if we can't write cache 

66 pass 

67 

68 def _fetch_latest_version(self) -> Optional[str]: 

69 """Fetch latest version from PyPI. 

70 

71 Returns: 

72 Latest version string or None if fetch failed 

73 """ 

74 try: 

75 # Create request with user agent 

76 req = Request( 

77 self.PYPI_API_URL, 

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

79 ) 

80 

81 # Set timeout to avoid hanging 

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

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

84 return data["info"]["version"] 

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

86 return None 

87 

88 def check_for_updates(self, silent: bool = False) -> Tuple[bool, Optional[str]]: 

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

90 

91 Args: 

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

93 

94 Returns: 

95 Tuple of (has_update, latest_version) 

96 """ 

97 # Skip check if current version is unknown 

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

99 return False, None 

100 

101 # Try to get from cache first 

102 cache_data = self._get_cache_data() 

103 if cache_data: 

104 latest_version = cache_data.get("latest_version") 

105 else: 

106 # Fetch from PyPI 

107 latest_version = self._fetch_latest_version() 

108 if latest_version: 

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

110 

111 if not latest_version: 

112 return False, None 

113 

114 try: 

115 # Compare versions 

116 current_ver = version.parse(self.current_version) 

117 latest_ver = version.parse(latest_version) 

118 

119 has_update = latest_ver > current_ver 

120 return has_update, latest_version 

121 except version.InvalidVersion: 

122 return False, None 

123 

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

125 """Show update notification to user. 

126 

127 Args: 

128 latest_version: Latest available version 

129 """ 

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

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

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

133 print( 

134 f" Update: pip install --upgrade qdrant-loader qdrant-loader-mcp-server" 

135 ) 

136 print() 

137 

138 

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

140 """Asynchronously check for version updates. 

141 

142 Args: 

143 current_version: Current version of qdrant-loader 

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

145 """ 

146 

147 def _check(): 

148 checker = VersionChecker(current_version) 

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

150 

151 if has_update and latest_version and not silent: 

152 checker.show_update_notification(latest_version) 

153 

154 # Run check in background thread to avoid blocking CLI 

155 try: 

156 import threading 

157 

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

159 thread.start() 

160 except Exception: 

161 # Silently fail if threading doesn't work 

162 pass