Coverage for src / tracekit / plugins / registry.py: 95%
115 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Plugin registry and management.
3This module provides the central plugin registry for loading,
4registering, and accessing plugins.
7Example:
8 >>> from tracekit.plugins.registry import register_plugin, get_plugin
9 >>> register_plugin(MyDecoder)
10 >>> decoder = get_plugin("my_decoder")
11"""
13from __future__ import annotations
15import logging
16from typing import TYPE_CHECKING, Any
18from tracekit.plugins.discovery import (
19 TRACEKIT_API_VERSION,
20 DiscoveredPlugin,
21 discover_plugins,
22)
24if TYPE_CHECKING:
25 from tracekit.plugins.base import PluginBase, PluginCapability, PluginMetadata
27logger = logging.getLogger(__name__)
30class PluginConflictError(Exception):
31 """Plugin registration conflict.
33 Raised when registering a plugin with a name that already exists.
35 Attributes:
36 existing: Metadata of existing plugin.
37 new: Metadata of new plugin.
38 """
40 def __init__(
41 self,
42 message: str,
43 existing: PluginMetadata,
44 new: PluginMetadata,
45 ) -> None:
46 super().__init__(message)
47 self.existing = existing
48 self.new = new
51class PluginVersionError(Exception):
52 """Plugin version incompatibility.
54 Raised when a plugin is not compatible with the current API.
56 Attributes:
57 plugin_api_version: Plugin's required API version.
58 tracekit_api_version: Current TraceKit API version.
59 """
61 def __init__(
62 self,
63 message: str,
64 plugin_api_version: str,
65 tracekit_api_version: str,
66 ) -> None:
67 super().__init__(message)
68 self.plugin_api_version = plugin_api_version
69 self.tracekit_api_version = tracekit_api_version
72class PluginDependencyError(Exception):
73 """Plugin dependency not satisfied.
75 Attributes:
76 plugin: Plugin name that has unmet dependency.
77 dependency: Missing dependency name.
78 required_version: Required dependency version.
79 """
81 def __init__(
82 self,
83 message: str,
84 plugin: str,
85 dependency: str,
86 required_version: str,
87 ) -> None:
88 super().__init__(message)
89 self.plugin = plugin
90 self.dependency = dependency
91 self.required_version = required_version
94class PluginRegistry:
95 """Central registry for plugins.
97 Manages plugin registration, loading, and lookup.
99 Example:
100 >>> registry = PluginRegistry()
101 >>> registry.register(MyDecoder)
102 >>> plugin = registry.get("my_decoder")
103 """
105 def __init__(self) -> None:
106 """Initialize empty registry."""
107 self._plugins: dict[str, PluginBase] = {}
108 self._metadata: dict[str, PluginMetadata] = {}
109 self._by_capability: dict[PluginCapability, list[str]] = {}
110 self._discovered: list[DiscoveredPlugin] = []
112 def register(
113 self,
114 plugin: type[PluginBase] | PluginBase,
115 *,
116 check_compatibility: bool = True,
117 check_conflicts: bool = True,
118 config: dict[str, Any] | None = None,
119 ) -> None:
120 """Register a plugin with the registry.
122 Args:
123 plugin: Plugin class or instance to register.
124 check_compatibility: Verify API compatibility.
125 check_conflicts: Check for duplicate names.
126 config: Optional plugin configuration.
128 Raises:
129 PluginConflictError: If plugin name already registered.
130 PluginVersionError: If plugin is not compatible.
131 """
132 # Get or create instance
133 instance = plugin() if isinstance(plugin, type) else plugin
135 metadata = instance.metadata
137 # Check compatibility
138 if check_compatibility and not metadata.is_compatible_with(TRACEKIT_API_VERSION):
139 raise PluginVersionError(
140 f"Plugin '{metadata.name}' requires API v{metadata.api_version}, "
141 f"but TraceKit API is v{TRACEKIT_API_VERSION}",
142 plugin_api_version=metadata.api_version,
143 tracekit_api_version=TRACEKIT_API_VERSION,
144 )
146 # Check conflicts (PLUG-002: conflict detection for duplicate plugins)
147 if check_conflicts and metadata.name in self._plugins:
148 existing = self._metadata[metadata.name]
150 # Provide detailed conflict information
151 conflict_msg = (
152 f"Plugin '{metadata.name}' already registered:\n"
153 f" Existing: v{existing.version} at {existing.path}\n"
154 f" New: v{metadata.version}"
155 )
157 # Check if same version
158 if existing.version == metadata.version: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 conflict_msg += " (same version)"
161 raise PluginConflictError(
162 conflict_msg,
163 existing=existing,
164 new=metadata,
165 )
167 # Register
168 self._plugins[metadata.name] = instance
169 self._metadata[metadata.name] = metadata
171 # Index by capability
172 for cap in metadata.capabilities:
173 if cap not in self._by_capability:
174 self._by_capability[cap] = []
175 self._by_capability[cap].append(metadata.name)
177 # Configure and load
178 if config:
179 instance.on_configure(config)
181 instance.on_load()
183 logger.info(f"Registered plugin: {metadata.name} v{metadata.version}")
185 def unregister(self, name: str) -> None:
186 """Unregister a plugin.
188 Args:
189 name: Plugin name to unregister.
190 """
191 if name not in self._plugins:
192 return
194 instance = self._plugins[name]
195 metadata = self._metadata[name]
197 # Call unload hook
198 instance.on_unload()
200 # Remove from capability index
201 for cap in metadata.capabilities:
202 if cap in self._by_capability and name in self._by_capability[cap]: 202 ↛ 201line 202 didn't jump to line 201 because the condition on line 202 was always true
203 self._by_capability[cap].remove(name)
205 # Remove from registry
206 del self._plugins[name]
207 del self._metadata[name]
209 logger.info(f"Unregistered plugin: {name}")
211 def get(self, name: str) -> PluginBase | None:
212 """Get plugin by name.
214 Args:
215 name: Plugin name.
217 Returns:
218 Plugin instance or None.
219 """
220 return self._plugins.get(name)
222 def get_metadata(self, name: str) -> PluginMetadata | None:
223 """Get plugin metadata by name.
225 Args:
226 name: Plugin name.
228 Returns:
229 Plugin metadata or None.
230 """
231 return self._metadata.get(name)
233 def list_plugins(
234 self,
235 *,
236 capability: PluginCapability | None = None,
237 ) -> list[PluginMetadata]:
238 """List registered plugins.
240 Args:
241 capability: Filter by capability.
243 Returns:
244 List of plugin metadata.
245 """
246 if capability is not None:
247 names = self._by_capability.get(capability, [])
248 return [self._metadata[name] for name in names]
250 return list(self._metadata.values())
252 def has_plugin(self, name: str) -> bool:
253 """Check if plugin is registered.
255 Args:
256 name: Plugin name.
258 Returns:
259 True if registered.
260 """
261 return name in self._plugins
263 def is_compatible(self, name: str) -> bool:
264 """Check if plugin is compatible with current API.
266 Args:
267 name: Plugin name.
269 Returns:
270 True if compatible.
271 """
272 metadata = self._metadata.get(name)
273 if metadata is None:
274 return False
275 return metadata.is_compatible_with(TRACEKIT_API_VERSION)
277 def discover_and_load(
278 self,
279 *,
280 compatible_only: bool = True,
281 config: dict[str, dict[str, Any]] | None = None,
282 ) -> list[PluginMetadata]:
283 """Discover and load all available plugins.
285 Args:
286 compatible_only: Only load compatible plugins.
287 config: Configuration dict keyed by plugin name.
289 Returns:
290 List of loaded plugin metadata.
291 """
292 self._discovered = discover_plugins(compatible_only=compatible_only)
293 loaded: list[PluginMetadata] = []
295 for discovered in self._discovered:
296 if discovered.load_error:
297 logger.warning(
298 f"Skipping plugin {discovered.metadata.name}: {discovered.load_error}"
299 )
300 continue
302 if not discovered.compatible and compatible_only:
303 logger.debug(f"Skipping incompatible plugin: {discovered.metadata.name}")
304 continue
306 try:
307 if config and discovered.metadata.name in config: 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true
308 config[discovered.metadata.name]
310 # For now, just store the metadata
311 # Full loading requires importing the plugin module
312 self._metadata[discovered.metadata.name] = discovered.metadata
313 loaded.append(discovered.metadata)
315 except Exception as e:
316 logger.error(f"Failed to load plugin {discovered.metadata.name}: {e}")
318 return loaded
320 def get_providers(self, item_type: str, item_name: str) -> list[str]:
321 """Find plugins that provide a specific capability.
323 Args:
324 item_type: Type of item (e.g., "protocols", "algorithms").
325 item_name: Name of item.
327 Returns:
328 List of plugin names that provide the item.
329 """
330 providers: list[str] = []
332 for name, metadata in self._metadata.items():
333 if item_type in metadata.provides: 333 ↛ 332line 333 didn't jump to line 332 because the condition on line 333 was always true
334 if item_name in metadata.provides[item_type]:
335 providers.append(name)
337 return providers
340# Global registry instance
341_global_registry: PluginRegistry | None = None
344def get_plugin_registry() -> PluginRegistry:
345 """Get the global plugin registry.
347 Returns:
348 Global PluginRegistry instance.
349 """
350 global _global_registry
352 if _global_registry is None:
353 _global_registry = PluginRegistry()
355 return _global_registry
358def register_plugin(
359 plugin: type[PluginBase] | PluginBase,
360 *,
361 config: dict[str, Any] | None = None,
362) -> None:
363 """Register plugin with global registry.
365 Args:
366 plugin: Plugin class or instance.
367 config: Plugin configuration.
368 """
369 get_plugin_registry().register(plugin, config=config)
372def get_plugin(name: str) -> PluginBase | None:
373 """Get plugin from global registry.
375 Args:
376 name: Plugin name.
378 Returns:
379 Plugin instance or None.
380 """
381 return get_plugin_registry().get(name)
384def list_plugins(
385 *,
386 capability: PluginCapability | None = None,
387) -> list[PluginMetadata]:
388 """List plugins from global registry.
390 Args:
391 capability: Filter by capability.
393 Returns:
394 List of plugin metadata.
395 """
396 return get_plugin_registry().list_plugins(capability=capability)
399def is_compatible(name: str) -> bool:
400 """Check plugin compatibility.
402 Args:
403 name: Plugin name.
405 Returns:
406 True if compatible.
407 """
408 return get_plugin_registry().is_compatible(name)
411__all__ = [
412 "PluginConflictError",
413 "PluginDependencyError",
414 "PluginRegistry",
415 "PluginVersionError",
416 "get_plugin",
417 "get_plugin_registry",
418 "is_compatible",
419 "list_plugins",
420 "register_plugin",
421]