Coverage for src / tracekit / plugins / discovery.py: 90%

1"""Plugin discovery and scanning. 

2 

3This module provides plugin discovery from filesystem directories 

4and Python entry points. 

5 

6 

7Example: 

8 >>> from tracekit.plugins.discovery import discover_plugins 

9 >>> plugins = discover_plugins() 

10 >>> for plugin in plugins: 

11 ... print(f"Found: {plugin.name} v{plugin.version}") 

12""" 

13 

14from __future__ import annotations 

15 

16import importlib 

17import importlib.metadata 

18import importlib.util 

19import os 

20import sys 

21from dataclasses import dataclass 

22from pathlib import Path 

23from typing import TYPE_CHECKING 

24 

25from tracekit.plugins.base import PluginBase, PluginMetadata 

26 

27if TYPE_CHECKING: 

28 from collections.abc import Iterator 

29 

30# Try to import yaml for plugin.yaml parsing 

31try: 

32 import yaml 

33 

34 YAML_AVAILABLE = True 

35except ImportError: 

36 YAML_AVAILABLE = False 

37 

38 

39# TraceKit API version for compatibility checking 

40TRACEKIT_API_VERSION = "1.0.0" 

41 

42 

43@dataclass 

44class DiscoveredPlugin: 

45 """Information about a discovered plugin. 

46 

47 Attributes: 

48 metadata: Plugin metadata. 

49 path: Path to plugin directory or module. 

50 entry_point: Entry point name (if from entry points). 

51 compatible: Whether plugin is compatible with current API. 

52 load_error: Error message if plugin failed to load. 

53 """ 

54 

55 metadata: PluginMetadata 

56 path: Path | None = None 

57 entry_point: str | None = None 

58 compatible: bool = True 

59 load_error: str | None = None 

60 

61 

62def get_plugin_paths() -> list[Path]: 

63 """Get list of plugin search directories. 

64 

65 Returns paths in priority order: 

66 1. Project plugins: ./plugins/ 

67 2. User plugins: ~/.tracekit/plugins/ 

68 3. System plugins: /usr/lib/tracekit/plugins/ 

69 

70 Returns: 

71 List of plugin directory paths. 

72 """ 

73 paths: list[Path] = [] 

74 

75 # Project plugins (current directory) 

76 project_plugins = Path.cwd() / "plugins" 

77 if project_plugins.exists(): 

78 paths.append(project_plugins) 

79 

80 # User plugins 

81 user_plugins = Path.home() / ".tracekit" / "plugins" 

82 paths.append(user_plugins) # Include even if doesn't exist 

83 

84 # XDG config home 

85 xdg_config = os.environ.get("XDG_CONFIG_HOME") 

86 if xdg_config: 

87 xdg_plugins = Path(xdg_config) / "tracekit" / "plugins" 

88 paths.append(xdg_plugins) 

89 

90 # System plugins (Linux) 

91 system_plugins = Path("/usr/lib/tracekit/plugins") 

92 if system_plugins.exists(): 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 paths.append(system_plugins) 

94 

95 # Local lib (Linux) 

96 local_plugins = Path("/usr/local/lib/tracekit/plugins") 

97 if local_plugins.exists(): 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true

98 paths.append(local_plugins) 

99 

100 return paths 

101 

102 

103def discover_plugins( 

104 *, 

105 compatible_only: bool = False, 

106 include_disabled: bool = False, 

107) -> list[DiscoveredPlugin]: 

108 """Discover all available plugins. 

109 

110 Scans plugin directories and Python entry points for plugins. 

111 

112 Args: 

113 compatible_only: If True, only return compatible plugins. 

114 include_disabled: If True, include disabled plugins. 

115 

116 Returns: 

117 List of discovered plugins. 

118 

119 Example: 

120 >>> plugins = discover_plugins() 

121 >>> print(f"Found {len(plugins)} plugins") 

122 """ 

123 plugins: list[DiscoveredPlugin] = [] 

124 seen_names: set[str] = set() 

125 

126 # Scan plugin directories 

127 for plugin_dir in get_plugin_paths(): 

128 if plugin_dir.exists() and plugin_dir.is_dir(): 128 ↛ 127line 128 didn't jump to line 127 because the condition on line 128 was always true

129 for plugin in scan_directory(plugin_dir): 

130 if plugin.metadata.name not in seen_names: 130 ↛ 129line 130 didn't jump to line 129 because the condition on line 130 was always true

131 plugins.append(plugin) 

132 seen_names.add(plugin.metadata.name) 

133 

134 # Scan entry points 

135 for plugin in scan_entry_points(): 

136 if plugin.metadata.name not in seen_names: 

137 plugins.append(plugin) 

138 seen_names.add(plugin.metadata.name) 

139 

140 # Filter by compatibility 

141 if compatible_only: 

142 plugins = [p for p in plugins if p.compatible] 

143 

144 # Filter disabled 

145 if not include_disabled: 

146 plugins = [p for p in plugins if p.metadata.enabled] 

147 

148 return plugins 

149 

150 

151def scan_directory(directory: Path) -> Iterator[DiscoveredPlugin]: 

152 """Scan a directory for plugins. 

153 

154 Each subdirectory with a plugin.yaml or Python package 

155 is considered a potential plugin. 

156 

157 Args: 

158 directory: Directory to scan. 

159 

160 Yields: 

161 DiscoveredPlugin for each found plugin. 

162 """ 

163 if not directory.exists(): 

164 return 

165 

166 for item in directory.iterdir(): 

167 if item.is_dir(): 

168 # Check for plugin.yaml 

169 plugin_yaml = item / "plugin.yaml" 

170 plugin_yml = item / "plugin.yml" 

171 

172 if plugin_yaml.exists(): 

173 plugin = _load_plugin_from_yaml(plugin_yaml) 

174 if plugin: 174 ↛ 182line 174 didn't jump to line 182 because the condition on line 174 was always true

175 yield plugin 

176 elif plugin_yml.exists(): 

177 plugin = _load_plugin_from_yaml(plugin_yml) 

178 if plugin: 178 ↛ 182line 178 didn't jump to line 182 because the condition on line 178 was always true

179 yield plugin 

180 

181 # Check for Python package with __init__.py 

182 init_py = item / "__init__.py" 

183 if init_py.exists(): 

184 plugin = _load_plugin_from_module(item) 

185 if plugin: 

186 yield plugin 

187 

188 

189def scan_entry_points() -> Iterator[DiscoveredPlugin]: 

190 """Scan Python entry points for plugins. 

191 

192 Looks for entry points in the "tracekit.plugins" group. 

193 

194 Yields: 

195 DiscoveredPlugin for each found entry point. 

196 """ 

197 try: 

198 # Python 3.10+ has entry_points with group filtering 

199 if hasattr(importlib.metadata, "entry_points"): 199 ↛ exitline 199 didn't return from function 'scan_entry_points' because the condition on line 199 was always true

200 eps = importlib.metadata.entry_points() 

201 

202 # Handle different API versions 

203 if hasattr(eps, "select"): 

204 # Python 3.10+ 

205 plugins_eps = eps.select(group="tracekit.plugins") 

206 elif hasattr(eps, "get"): 206 ↛ 211line 206 didn't jump to line 211 because the condition on line 206 was always true

207 # Python 3.9 

208 plugins_eps = eps.get("tracekit.plugins", []) 

209 else: 

210 # Python 3.8 style (dict) 

211 plugins_eps = eps.get("tracekit.plugins", []) # type: ignore[attr-defined] 

212 

213 for ep in plugins_eps: 

214 try: 

215 plugin_class = ep.load() 

216 

217 if isinstance(plugin_class, type) and issubclass(plugin_class, PluginBase): 

218 instance = plugin_class() 

219 metadata = instance.metadata 

220 

221 compatible = metadata.is_compatible_with(TRACEKIT_API_VERSION) 

222 

223 yield DiscoveredPlugin( 

224 metadata=metadata, 

225 entry_point=ep.name, 

226 compatible=compatible, 

227 ) 

228 except Exception as e: 

229 # Create placeholder for failed load 

230 yield DiscoveredPlugin( 

231 metadata=PluginMetadata( 

232 name=ep.name, 

233 version="0.0.0", 

234 description="Failed to load", 

235 ), 

236 entry_point=ep.name, 

237 compatible=False, 

238 load_error=str(e), 

239 ) 

240 

241 except Exception: 

242 # Entry points not available or error 

243 pass 

244 

245 

246def _load_plugin_from_yaml(yaml_path: Path) -> DiscoveredPlugin | None: 

247 """Load plugin metadata from YAML file. 

248 

249 Args: 

250 yaml_path: Path to plugin.yaml file. 

251 

252 Returns: 

253 DiscoveredPlugin or None if load fails. 

254 """ 

255 if not YAML_AVAILABLE: 

256 return None 

257 

258 try: 

259 with open(yaml_path, encoding="utf-8") as f: 

260 data = yaml.safe_load(f) 

261 

262 if not isinstance(data, dict): 

263 return None 

264 

265 # Extract metadata 

266 metadata = PluginMetadata( 

267 name=data.get("name", yaml_path.parent.name), 

268 version=data.get("version", "0.0.0"), 

269 api_version=data.get("api_version", "1.0.0"), 

270 author=data.get("author", ""), 

271 description=data.get("description", ""), 

272 homepage=data.get("homepage", ""), 

273 license=data.get("license", ""), 

274 path=yaml_path.parent, 

275 enabled=data.get("enabled", True), 

276 ) 

277 

278 # Parse dependencies 

279 if "dependencies" in data: 

280 deps = data["dependencies"] 

281 if isinstance(deps, list): 281 ↛ 290line 281 didn't jump to line 290 because the condition on line 281 was always true

282 for dep in deps: 

283 if isinstance(dep, dict): 283 ↛ 282line 283 didn't jump to line 282 because the condition on line 283 was always true

284 if "plugin" in dep: 

285 metadata.dependencies[dep["plugin"]] = dep.get("version", "*") 

286 elif "package" in dep: 286 ↛ 282line 286 didn't jump to line 282 because the condition on line 286 was always true

287 metadata.dependencies[dep["package"]] = dep.get("version", "*") 

288 

289 # Parse provides 

290 if "provides" in data: 

291 provides = data["provides"] 

292 if isinstance(provides, list): 292 ↛ 300line 292 didn't jump to line 300 because the condition on line 292 was always true

293 for item in provides: 

294 if isinstance(item, dict): 294 ↛ 293line 294 didn't jump to line 293 because the condition on line 294 was always true

295 for key, value in item.items(): 

296 if key not in metadata.provides: 

297 metadata.provides[key] = [] 

298 metadata.provides[key].append(value) 

299 

300 compatible = metadata.is_compatible_with(TRACEKIT_API_VERSION) 

301 

302 return DiscoveredPlugin( 

303 metadata=metadata, 

304 path=yaml_path.parent, 

305 compatible=compatible, 

306 ) 

307 

308 except Exception as e: 

309 return DiscoveredPlugin( 

310 metadata=PluginMetadata( 

311 name=yaml_path.parent.name, 

312 version="0.0.0", 

313 path=yaml_path.parent, 

314 ), 

315 path=yaml_path.parent, 

316 compatible=False, 

317 load_error=str(e), 

318 ) 

319 

320 

321def _load_plugin_from_module(module_path: Path) -> DiscoveredPlugin | None: 

322 """Load plugin from Python module. 

323 

324 Args: 

325 module_path: Path to Python package directory. 

326 

327 Returns: 

328 DiscoveredPlugin or None if load fails. 

329 """ 

330 try: 

331 # Add parent to path temporarily 

332 parent = str(module_path.parent) 

333 if parent not in sys.path: 333 ↛ 337line 333 didn't jump to line 337 because the condition on line 333 was always true

334 sys.path.insert(0, parent) 

335 added_path = True 

336 else: 

337 added_path = False 

338 

339 try: 

340 # Import the module 

341 module_name = module_path.name 

342 spec = importlib.util.spec_from_file_location(module_name, module_path / "__init__.py") 

343 

344 if spec is None or spec.loader is None: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true

345 return None 

346 

347 module = importlib.util.module_from_spec(spec) 

348 spec.loader.exec_module(module) 

349 

350 # Look for Plugin class 

351 plugin_class = None 

352 

353 # Check for explicit Plugin class 

354 if hasattr(module, "Plugin"): 

355 plugin_class = module.Plugin 

356 elif hasattr(module, "plugin"): 356 ↛ 357line 356 didn't jump to line 357 because the condition on line 356 was never true

357 plugin_class = module.plugin 

358 

359 # Check for any PluginBase subclass 

360 if plugin_class is None: 

361 for attr_name in dir(module): 

362 attr = getattr(module, attr_name) 

363 if ( 363 ↛ 368line 363 didn't jump to line 368 because the condition on line 363 was never true

364 isinstance(attr, type) 

365 and issubclass(attr, PluginBase) 

366 and attr is not PluginBase 

367 ): 

368 plugin_class = attr 

369 break 

370 

371 if plugin_class is None: 

372 return None 

373 

374 # Create instance and get metadata 

375 instance = plugin_class() 

376 metadata = instance.metadata 

377 metadata.path = module_path 

378 

379 compatible = metadata.is_compatible_with(TRACEKIT_API_VERSION) 

380 

381 return DiscoveredPlugin( 

382 metadata=metadata, 

383 path=module_path, 

384 compatible=compatible, 

385 ) 

386 

387 finally: 

388 if added_path: 

389 sys.path.remove(parent) 

390 

391 except Exception as e: 

392 return DiscoveredPlugin( 

393 metadata=PluginMetadata( 

394 name=module_path.name, 

395 version="0.0.0", 

396 path=module_path, 

397 ), 

398 path=module_path, 

399 compatible=False, 

400 load_error=str(e), 

401 ) 

402 

403 

404__all__ = [ 

405 "TRACEKIT_API_VERSION", 

406 "DiscoveredPlugin", 

407 "discover_plugins", 

408 "get_plugin_paths", 

409 "scan_directory", 

410 "scan_entry_points", 

411]