Coverage for lintro / plugins / registry.py: 96%

1"""Tool registry for discovering and managing Lintro plugins. 

2 

3This module provides a central registry for all Lintro tools, supporting 

4both built-in tools and external plugins discovered via entry points. 

5 

6Example: 

7 >>> from lintro.plugins.registry import ToolRegistry, register_tool 

8 >>> from lintro.plugins.base import BaseToolPlugin 

9 >>> 

10 >>> @register_tool 

11 ... class MyPlugin(BaseToolPlugin): 

12 ... # Plugin implementation 

13 ... pass 

14 >>> 

15 >>> tool = ToolRegistry.get("my-tool") 

16""" 

17 

18from __future__ import annotations 

19 

20import threading 

21from typing import TYPE_CHECKING 

22 

23from loguru import logger 

24 

25from lintro.plugins.base import BaseToolPlugin 

26 

27if TYPE_CHECKING: 

28 from lintro.plugins.protocol import ToolDefinition 

29 

30 

31class ToolRegistry: 

32 """Central registry for all Lintro tools. 

33 

34 This class maintains a registry of all available tools, both built-in 

35 and external plugins. It provides methods for registering, retrieving, 

36 and listing tools. 

37 

38 The registry is thread-safe and uses lazy instantiation for tool instances. 

39 """ 

40 

41 _tools: dict[str, type[BaseToolPlugin]] = {} 

42 _instances: dict[str, BaseToolPlugin] = {} 

43 _lock: threading.RLock = threading.RLock() # Reentrant lock for nested calls 

44 

45 @classmethod 

46 def register(cls, plugin_class: type[BaseToolPlugin]) -> type[BaseToolPlugin]: 

47 """Register a tool class. 

48 

49 Can be used as a decorator or called directly. 

50 

51 Args: 

52 plugin_class: The tool class to register. 

53 

54 Returns: 

55 The registered tool class (allows use as decorator). 

56 

57 Example: 

58 >>> @ToolRegistry.register 

59 ... class MyPlugin(BaseToolPlugin): 

60 ... pass 

61 """ 

62 with cls._lock: 

63 # Create a temporary instance to get the definition 

64 instance = plugin_class() 

65 name = instance.definition.name.lower() 

66 

67 if name in cls._tools: 

68 existing = cls._tools[name] 

69 logger.warning( 

70 f"Tool '{name}' already registered by {existing.__module__}." 

71 f"{existing.__name__}, overwriting with {plugin_class.__module__}." 

72 f"{plugin_class.__name__}", 

73 ) 

74 

75 cls._tools[name] = plugin_class 

76 # Store the instance we created for get() calls 

77 cls._instances[name] = instance 

78 logger.debug(f"Registered tool: {name}") 

79 

80 return plugin_class 

81 

82 @classmethod 

83 def _ensure_discovered(cls) -> None: 

84 """Ensure tools have been discovered. 

85 

86 This is called automatically when accessing tools to support 

87 lazy discovery when the package is imported. 

88 """ 

89 if not cls._tools: 

90 # Auto-discover tools if registry is empty 

91 from lintro.plugins.discovery import discover_all_tools 

92 

93 discover_all_tools() 

94 

95 @classmethod 

96 def get(cls, name: str) -> BaseToolPlugin: 

97 """Get a tool instance by name. 

98 

99 Args: 

100 name: Tool name (case-insensitive). 

101 

102 Returns: 

103 The tool instance. 

104 

105 Raises: 

106 ValueError: If the tool is not registered. 

107 

108 Example: 

109 >>> tool = ToolRegistry.get("hadolint") 

110 >>> result = tool.check(["."], {}) 

111 """ 

112 name_lower = name.lower() 

113 

114 with cls._lock: 

115 # Auto-discover tools if not yet done 

116 cls._ensure_discovered() 

117 

118 if name_lower not in cls._instances: 

119 if name_lower not in cls._tools: 

120 available = ", ".join(sorted(cls._tools.keys())) 

121 raise ValueError( 

122 f"Unknown tool: {name!r}. " 

123 f"Available tools: {available or 'none'}", 

124 ) 

125 cls._instances[name_lower] = cls._tools[name_lower]() 

126 

127 return cls._instances[name_lower] 

128 

129 @classmethod 

130 def get_all(cls) -> dict[str, BaseToolPlugin]: 

131 """Get all registered tool instances. 

132 

133 Returns: 

134 Dictionary mapping tool names to tool instances. 

135 

136 Example: 

137 >>> all_tools = ToolRegistry.get_all() 

138 >>> for name, tool in all_tools.items(): 

139 ... print(f"{name}: {tool.definition.description}") 

140 """ 

141 with cls._lock: 

142 cls._ensure_discovered() 

143 return {name: cls.get(name) for name in cls._tools} 

144 

145 @classmethod 

146 def get_definitions(cls) -> dict[str, ToolDefinition]: 

147 """Get all tool definitions. 

148 

149 Returns: 

150 Dictionary mapping tool names to their definitions. 

151 

152 Example: 

153 >>> defs = ToolRegistry.get_definitions() 

154 >>> for name, defn in defs.items(): 

155 ... print(f"{name}: can_fix={defn.can_fix}") 

156 """ 

157 with cls._lock: 

158 cls._ensure_discovered() 

159 return {name: cls.get(name).definition for name in cls._tools} 

160 

161 @classmethod 

162 def get_names(cls) -> list[str]: 

163 """Get all registered tool names. 

164 

165 Returns: 

166 Sorted list of tool names. 

167 """ 

168 with cls._lock: 

169 cls._ensure_discovered() 

170 return sorted(cls._tools.keys()) 

171 

172 @classmethod 

173 def is_registered(cls, name: str) -> bool: 

174 """Check if a tool is registered. 

175 

176 Args: 

177 name: Tool name (case-insensitive). 

178 

179 Returns: 

180 True if the tool is registered, False otherwise. 

181 """ 

182 with cls._lock: 

183 cls._ensure_discovered() 

184 return name.lower() in cls._tools 

185 

186 @classmethod 

187 def clear(cls) -> None: 

188 """Clear all registered tools. 

189 

190 This is primarily useful for testing. 

191 """ 

192 with cls._lock: 

193 cls._tools.clear() 

194 cls._instances.clear() 

195 logger.debug("Cleared tool registry") 

196 

197 @classmethod 

198 def get_check_tools(cls) -> dict[str, BaseToolPlugin]: 

199 """Get all tools that support checking (all tools). 

200 

201 Returns: 

202 Dictionary mapping tool names to tool instances. 

203 """ 

204 return cls.get_all() 

205 

206 @classmethod 

207 def get_fix_tools(cls) -> dict[str, BaseToolPlugin]: 

208 """Get all tools that support fixing. 

209 

210 Returns: 

211 Dictionary mapping tool names to tool instances for fix-capable tools. 

212 """ 

213 all_tools = cls.get_all() 

214 return { 

215 name: tool for name, tool in all_tools.items() if tool.definition.can_fix 

216 } 

217 

218 

219def register_tool(cls: type[BaseToolPlugin]) -> type[BaseToolPlugin]: 

220 """Decorator to register a tool class. 

221 

222 This is a convenience function that wraps ToolRegistry.register(). 

223 

224 Args: 

225 cls: The tool class to register. 

226 

227 Returns: 

228 The registered tool class. 

229 

230 Example: 

231 >>> from lintro.plugins.registry import register_tool 

232 >>> from lintro.plugins.base import BaseToolPlugin 

233 >>> 

234 >>> @register_tool 

235 ... class HadolintPlugin(BaseToolPlugin): 

236 ... @property 

237 ... def definition(self) -> ToolDefinition: 

238 ... return ToolDefinition(name="hadolint", description="...") 

239 """ 

240 return ToolRegistry.register(cls)