Coverage for lintro / plugins / discovery.py: 87%

71 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-04-03 18:53 +0000

1"""Tool discovery for builtin and external plugins. 

2 

3This module handles discovering and loading Lintro tools from: 

41. Built-in tool definitions (lintro/tools/definitions/) 

52. External plugins via Python entry points (lintro.plugins) 

6 

7Example: 

8 >>> from lintro.plugins.discovery import discover_all_tools 

9 >>> discover_all_tools() # Loads all available tools 

10""" 

11 

12from __future__ import annotations 

13 

14import importlib 

15import importlib.metadata 

16from pathlib import Path 

17from typing import cast 

18 

19from loguru import logger 

20 

21from lintro.plugins.base import BaseToolPlugin 

22from lintro.plugins.registry import ToolRegistry 

23 

24# Path to builtin tool definitions 

25BUILTIN_DEFINITIONS_PATH = Path(__file__).parent.parent / "tools" / "definitions" 

26 

27# Entry point group for external plugins 

28ENTRY_POINT_GROUP = "lintro.plugins" 

29 

30# Track whether discovery has been performed 

31_discovered: bool = False 

32 

33 

34def discover_builtin_tools() -> int: 

35 """Load all builtin tool definitions. 

36 

37 This function imports all Python modules in the tools/definitions/ 

38 directory, which triggers the @register_tool decorators. 

39 

40 Returns: 

41 Number of tool modules loaded. 

42 

43 Note: 

44 Each tool definition file should use the @register_tool decorator 

45 to register itself with the ToolRegistry. 

46 """ 

47 loaded_count = 0 

48 

49 if not BUILTIN_DEFINITIONS_PATH.exists(): 

50 logger.warning( 

51 f"Builtin definitions path not found: {BUILTIN_DEFINITIONS_PATH}", 

52 ) 

53 return loaded_count 

54 

55 for py_file in BUILTIN_DEFINITIONS_PATH.glob("*.py"): 

56 if py_file.name.startswith("_"): 

57 continue 

58 

59 module_name = f"lintro.tools.definitions.{py_file.stem}" 

60 try: 

61 # Safe: module_name from internal directory files, not user input 

62 importlib.import_module(module_name) # nosemgrep: non-literal-import 

63 logger.debug(f"Loaded builtin tool: {py_file.stem}") 

64 loaded_count += 1 

65 except ImportError as e: 

66 logger.warning(f"Failed to import {module_name}: {e}") 

67 except (AttributeError, TypeError, ValueError) as e: 

68 logger.error(f"Error loading {module_name}: {type(e).__name__}: {e}") 

69 

70 logger.debug(f"Loaded {loaded_count} builtin tool definitions") 

71 return loaded_count 

72 

73 

74def discover_external_plugins() -> int: 

75 """Load external plugins via entry points. 

76 

77 External plugins can register themselves by defining an entry point 

78 in their pyproject.toml or setup.py: 

79 

80 [project.entry-points."lintro.plugins"] 

81 my-tool = "my_package.plugin:MyToolPlugin" 

82 

83 Returns: 

84 Number of external plugins loaded. 

85 

86 Note: 

87 External plugins should be classes that implement LintroPlugin. 

88 They will be automatically registered with the ToolRegistry. 

89 """ 

90 loaded_count = 0 

91 

92 try: 

93 entry_points = importlib.metadata.entry_points(group=ENTRY_POINT_GROUP) 

94 except (TypeError, AttributeError, KeyError) as e: 

95 logger.debug(f"No entry points found or error accessing them: {e}") 

96 return loaded_count 

97 

98 for ep in entry_points: 

99 try: 

100 plugin_class = ep.load() 

101 

102 # Validate that it's a proper plugin class 

103 if not isinstance(plugin_class, type): 

104 logger.warning( 

105 f"Entry point {ep.name!r} does not point to a class, skipping", 

106 ) 

107 continue 

108 

109 # Check if it implements LintroPlugin protocol (without instantiating) 

110 # Check for required attributes since Protocol with properties 

111 # can't use issubclass reliably 

112 required_attrs = ("definition", "check", "fix", "set_options") 

113 if not all(hasattr(plugin_class, attr) for attr in required_attrs): 

114 logger.warning( 

115 f"Entry point {ep.name!r} class does not implement LintroPlugin, " 

116 "skipping", 

117 ) 

118 continue 

119 

120 # Register the plugin if not already registered 

121 if not ToolRegistry.is_registered(ep.name): 

122 ToolRegistry.register(cast(type[BaseToolPlugin], plugin_class)) 

123 logger.info(f"Loaded external plugin: {ep.name}") 

124 loaded_count += 1 

125 else: 

126 logger.debug(f"Plugin {ep.name!r} already registered, skipping") 

127 

128 except (ImportError, AttributeError, TypeError, RuntimeError) as e: 

129 logger.warning(f"Failed to load plugin {ep.name!r}: {e}") 

130 

131 logger.debug(f"Loaded {loaded_count} external plugins") 

132 return loaded_count 

133 

134 

135def discover_all_tools(force: bool = False) -> int: 

136 """Discover and register all available tools. 

137 

138 This function loads both builtin tools and external plugins. 

139 It's safe to call multiple times - subsequent calls are no-ops 

140 unless force=True. 

141 

142 Args: 

143 force: If True, re-discover even if already discovered. 

144 

145 Returns: 

146 Total number of tools loaded. 

147 

148 Example: 

149 >>> from lintro.plugins.discovery import discover_all_tools 

150 >>> count = discover_all_tools() 

151 >>> print(f"Loaded {count} tools") 

152 """ 

153 global _discovered 

154 

155 if _discovered and not force: 

156 logger.debug("Tools already discovered, skipping") 

157 return 0 

158 

159 logger.debug("Discovering tools...") 

160 

161 # Discover builtin tools first 

162 builtin_count = discover_builtin_tools() 

163 

164 # Then discover external plugins (skips already-registered tool names) 

165 external_count = discover_external_plugins() 

166 

167 total = builtin_count + external_count 

168 _discovered = True 

169 

170 logger.info( 

171 f"Tool discovery complete: {builtin_count} builtin, {external_count} external", 

172 ) 

173 return total 

174 

175 

176def is_discovered() -> bool: 

177 """Check if tool discovery has been performed. 

178 

179 Returns: 

180 True if discover_all_tools() has been called, False otherwise. 

181 """ 

182 return _discovered 

183 

184 

185def reset_discovery() -> None: 

186 """Reset the discovery state. 

187 

188 This is primarily useful for testing. 

189 """ 

190 global _discovered 

191 _discovered = False