Coverage for src/quickmcp/factory/import_analyzer.py: 87%

1""" 

2Import analysis for detecting missing dependencies. 

3""" 

4 

5import ast 

6import importlib 

7import logging 

8from dataclasses import dataclass 

9from pathlib import Path 

10from typing import List, Optional, Set, Dict, Any 

11 

12from .config import FactoryConfig, DEFAULT_CONFIG 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17@dataclass 

18class MissingDependency: 

19 """Information about a missing dependency.""" 

20 module: str 

21 import_type: str # "import", "from_import", "optional" 

22 line_number: Optional[int] = None 

23 source_line: Optional[str] = None 

24 suggested_install: Optional[str] = None 

25 is_dev_dependency: bool = False 

26 

27 

28class ImportAnalysisError(Exception): 

29 """Raised when import analysis fails.""" 

30 pass 

31 

32 

33class ImportAnalyzer: 

34 """Analyze Python files for imports and missing dependencies.""" 

35 

36 # Common package name to pip install name mappings 

37 BASE_PIP_MAPPINGS = { 

38 'aiohttp': 'aiohttp', 

39 'requests': 'requests', 

40 'fastapi': 'fastapi', 

41 'starlette': 'starlette', 

42 'uvicorn': 'uvicorn', 

43 'pydantic': 'pydantic', 

44 'numpy': 'numpy', 

45 'pandas': 'pandas', 

46 'sqlalchemy': 'sqlalchemy', 

47 'psycopg2': 'psycopg2-binary', 

48 'mysql': 'mysql-connector-python', 

49 'redis': 'redis', 

50 'celery': 'celery', 

51 'click': 'click', 

52 'typer': 'typer', 

53 'rich': 'rich', 

54 'httpx': 'httpx', 

55 'websockets': 'websockets', 

56 'jinja2': 'jinja2', 

57 'flask': 'flask', 

58 'django': 'django', 

59 'boto3': 'boto3', 

60 'openai': 'openai', 

61 'anthropic': 'anthropic', 

62 'langchain': 'langchain', 

63 'transformers': 'transformers', 

64 'torch': 'torch', 

65 'tensorflow': 'tensorflow', 

66 'sklearn': 'scikit-learn', 

67 'cv2': 'opencv-python', 

68 'PIL': 'Pillow', 

69 'yaml': 'PyYAML', 

70 'toml': 'toml', 

71 'dotenv': 'python-dotenv', 

72 'jwt': 'PyJWT', 

73 'bcrypt': 'bcrypt', 

74 'cryptography': 'cryptography', 

75 'bs4': 'beautifulsoup4', 

76 } 

77 

78 # Development/testing packages 

79 DEV_PACKAGES = { 

80 'pytest', 'mypy', 'black', 'flake8', 'pylint', 'coverage', 

81 'tox', 'pre-commit', 'sphinx', 'mkdocs', 'bandit', 

82 'isort', 'autopep8', 'yapf', 'ruff' 

83 } 

84 

85 # Standard library modules (Python 3.11+) 

86 BASE_STDLIB_MODULES = { 

87 'os', 'sys', 'json', 'ast', 'inspect', 'importlib', 'pathlib', 

88 'typing', 'logging', 'asyncio', 'traceback', 'dataclasses', 

89 'functools', 'itertools', 'collections', 'datetime', 'time', 

90 'math', 'random', 'string', 're', 'urllib', 'http', 'email', 

91 'xml', 'html', 'sqlite3', 'pickle', 'csv', 'configparser', 

92 'argparse', 'subprocess', 'threading', 'multiprocessing', 

93 'queue', 'socket', 'ssl', 'hashlib', 'uuid', 'tempfile', 

94 'shutil', 'glob', 'fnmatch', 'zipfile', 'tarfile', 'gzip', 

95 'io', 'contextlib', 'warnings', 'unittest', 'doctest', 

96 'operator', 'copy', 'weakref', 'gc', 'atexit', 'signal', 

97 'errno', 'stat', 'filecmp', 'linecache', 'tokenize', 

98 'keyword', 'heapq', 'bisect', 'array', 'struct', 'codecs' 

99 } 

100 

101 def __init__(self, config: Optional[FactoryConfig] = None): 

102 """Initialize the import analyzer.""" 

103 self.config = config or DEFAULT_CONFIG 

104 self._pip_mappings = self.config.merge_pip_mappings(self.BASE_PIP_MAPPINGS) 

105 self._stdlib_modules = self.config.get_effective_stdlib_modules(self.BASE_STDLIB_MODULES) 

106 self._analysis_cache: Dict[str, List[MissingDependency]] = {} 

107 

108 def analyze_file(self, file_path: str) -> List[MissingDependency]: 

109 """Analyze a Python file for missing dependencies.""" 

110 file_path = str(Path(file_path).resolve()) 

111 

112 # Check cache if enabled 

113 if self.config.cache_dependency_analysis and file_path in self._analysis_cache: 

114 logger.debug(f"Using cached analysis for {file_path}") 

115 return self._analysis_cache[file_path] 

116 

117 try: 

118 missing_deps = self._analyze_file_impl(file_path) 

119 

120 # Cache results if enabled 

121 if self.config.cache_dependency_analysis: 

122 self._analysis_cache[file_path] = missing_deps 

123 

124 return missing_deps 

125 

126 except Exception as e: 

127 error_msg = f"Failed to analyze imports in {file_path}: {e}" 

128 if self.config.verbose_errors: 

129 logger.error(error_msg, exc_info=True) 

130 else: 

131 logger.error(error_msg) 

132 raise ImportAnalysisError(error_msg) from e 

133 

134 def _analyze_file_impl(self, file_path: str) -> List[MissingDependency]: 

135 """Internal implementation of file analysis.""" 

136 path = Path(file_path) 

137 

138 # Check file size 

139 if path.stat().st_size > self.config.max_file_size_mb * 1024 * 1024: 

140 raise ImportAnalysisError(f"File {file_path} is too large (>{self.config.max_file_size_mb}MB)") 

141 

142 # Read and parse file 

143 try: 

144 with open(file_path, 'r', encoding='utf-8') as f: 

145 content = f.read() 

146 lines = content.splitlines() 

147 except UnicodeDecodeError: 

148 raise ImportAnalysisError(f"Cannot decode file {file_path} as UTF-8") 

149 except IOError as e: 

150 raise ImportAnalysisError(f"Cannot read file {file_path}: {e}") 

151 

152 # Parse AST 

153 try: 

154 tree = ast.parse(content, filename=file_path) 

155 except SyntaxError as e: 

156 logger.warning(f"Syntax error in {file_path}: {e}") 

157 # Return empty list for files with syntax errors 

158 return [] 

159 

160 # Find all import statements 

161 missing_deps = [] 

162 for node in ast.walk(tree): 

163 if isinstance(node, ast.Import): 

164 for alias in node.names: 

165 missing = self._check_module(alias.name, "import", node.lineno, lines) 

166 if missing: 

167 missing_deps.append(missing) 

168 

169 elif isinstance(node, ast.ImportFrom): 

170 # Skip relative imports (level > 0 means relative) 

171 if node.level > 0: 

172 logger.debug(f"Skipping relative import at line {node.lineno}") 

173 continue 

174 

175 if node.module: 

176 missing = self._check_module(node.module, "from_import", node.lineno, lines) 

177 if missing: 

178 missing_deps.append(missing) 

179 

180 # Remove duplicates while preserving order 

181 seen = set() 

182 unique_deps = [] 

183 for dep in missing_deps: 

184 key = (dep.module, dep.import_type, dep.line_number) 

185 if key not in seen: 

186 seen.add(key) 

187 unique_deps.append(dep) 

188 

189 return unique_deps 

190 

191 def _check_module(self, module_name: str, import_type: str, line_num: int, lines: List[str]) -> Optional[MissingDependency]: 

192 """Check if a module is available.""" 

193 # Get the top-level module name 

194 top_module = module_name.split('.')[0] 

195 

196 # Skip standard library modules 

197 if top_module in self._stdlib_modules: 

198 logger.debug(f"Skipping stdlib module: {top_module}") 

199 return None 

200 

201 try: 

202 importlib.import_module(top_module) 

203 logger.debug(f"Module {top_module} is available") 

204 return None # Module is available 

205 except ImportError: 

206 logger.debug(f"Module {top_module} is missing") 

207 pass 

208 except Exception as e: 

209 # Other import errors (circular imports, etc.) 

210 logger.warning(f"Unexpected error importing {top_module}: {e}") 

211 return None 

212 

213 # Get source line 

214 source_line = lines[line_num - 1] if 0 < line_num <= len(lines) else None 

215 

216 # Check if this is an optional import (in try/except block) 

217 is_optional = self._is_optional_import(lines, line_num - 1) 

218 

219 # Check if it's a development dependency 

220 is_dev = top_module in self.DEV_PACKAGES 

221 

222 return MissingDependency( 

223 module=top_module, 

224 import_type="optional" if is_optional else import_type, 

225 line_number=line_num, 

226 source_line=source_line.strip() if source_line else None, 

227 suggested_install=self._pip_mappings.get(top_module, top_module), 

228 is_dev_dependency=is_dev 

229 ) 

230 

231 def _is_optional_import(self, lines: List[str], line_index: int) -> bool: 

232 """Check if an import is inside a try/except block or TYPE_CHECKING block.""" 

233 if line_index < 0 or line_index >= len(lines): 

234 return False 

235 

236 # Get the indentation level of the import line 

237 import_line = lines[line_index] 

238 import_indent = len(import_line) - len(import_line.lstrip()) 

239 

240 # Check for TYPE_CHECKING block - must be indented under it 

241 in_type_checking = False 

242 for i in range(line_index - 1, max(-1, line_index - 10), -1): 

243 line = lines[i] 

244 if not line.strip(): # Skip empty lines 

245 continue 

246 

247 line_indent = len(line) - len(line.lstrip()) 

248 

249 # Check if TYPE_CHECKING block is above us 

250 if 'if TYPE_CHECKING:' in line and line_indent < import_indent: 

251 in_type_checking = True 

252 break 

253 

254 # If we find a line at same or lesser indentation, check if we're leaving the block 

255 if line_indent < import_indent and not line.strip().startswith(('import', 'from')): 

256 # We've left the import block 

257 break 

258 

259 if in_type_checking: 

260 return True 

261 

262 # Check if we're in an except block 

263 for i in range(line_index - 1, max(-1, line_index - 10), -1): 

264 line = lines[i] 

265 if not line.strip(): # Skip empty lines 

266 continue 

267 

268 line_indent = len(line) - len(line.lstrip()) 

269 

270 # If we find an except at same or lesser indentation, we're optional 

271 if line_indent < import_indent and line.strip().startswith('except'): 

272 return True 

273 

274 # Look backwards for 'try:' at any indentation less than import 

275 # This handles nested try blocks 

276 try_indent = None 

277 for i in range(line_index - 1, max(-1, line_index - 20), -1): 

278 line = lines[i] 

279 if not line.strip(): # Skip empty lines 

280 continue 

281 

282 line_indent = len(line) - len(line.lstrip()) 

283 

284 # Look for try: at any level less indented than the import 

285 if line_indent < import_indent and line.strip().startswith('try:'): 

286 try_indent = line_indent 

287 break 

288 

289 if try_indent is None: 

290 return False 

291 

292 # Look forwards for 'except' block at the same indentation as the found try 

293 for i in range(line_index + 1, min(len(lines), line_index + 20)): 

294 line = lines[i] 

295 if not line.strip(): # Skip empty lines 

296 continue 

297 

298 line_indent = len(line) - len(line.lstrip()) 

299 

300 # Look for except at the same level as the try 

301 if line_indent == try_indent: 

302 if line.strip().startswith('except'): 

303 return True 

304 # Don't stop on other code at try level, as except might come later 

305 

306 return False 

307 

308 def get_install_commands(self, missing_deps: List[MissingDependency], use_uv: bool = None) -> Dict[str, str]: 

309 """Get install commands for missing dependencies.""" 

310 import shutil 

311 

312 if use_uv is None: 

313 use_uv = shutil.which('uv') is not None 

314 

315 commands = {} 

316 for dep in missing_deps: 

317 if dep.import_type == "optional": 

318 key = f"optional:{dep.module}" 

319 else: 

320 key = dep.module 

321 

322 cmd = "uv pip install" if use_uv else "pip install" 

323 commands[key] = f"{cmd} {dep.suggested_install}" 

324 

325 return commands 

326 

327 def clear_cache(self): 

328 """Clear the analysis cache.""" 

329 self._analysis_cache.clear() 

330 logger.debug("Cleared import analysis cache") 

331 

332 def get_cache_info(self) -> Dict[str, Any]: 

333 """Get information about the cache.""" 

334 return { 

335 "enabled": self.config.cache_dependency_analysis, 

336 "entries": len(self._analysis_cache), 

337 "files": list(self._analysis_cache.keys()) 

338 }