Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/services/plugins/entrypoints.py: 100%

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Discovers and loads plugins distributed as Python packages. 

5 

6This module provides the `load_entrypoints` function, which is responsible for 

7finding and loading plugins that have been installed into the Python 

8environment and registered under the `bijux_cli.plugins` entry point group. 

9This enables a distributable plugin ecosystem where plugins can be managed 

10via tools like `pip`. 

11""" 

12 

13from __future__ import annotations 

14 

15import asyncio 

16import contextlib 

17import importlib.metadata as im 

18import logging 

19import traceback 

20from typing import Any 

21 

22from packaging.specifiers import SpecifierSet 

23from packaging.version import Version as PkgVersion 

24 

25from bijux_cli.contracts import ( 

26 ObservabilityProtocol, 

27 RegistryProtocol, 

28 TelemetryProtocol, 

29) 

30from bijux_cli.core.di import DIContainer 

31 

32_LOG = logging.getLogger("bijux_cli.plugin_loader") 

33 

34 

35def _iter_plugin_eps() -> list[im.EntryPoint]: 

36 """Returns all entry points in the 'bijux_cli.plugins' group. 

37 

38 Returns: 

39 list[importlib.metadata.EntryPoint]: A list of all discovered plugin 

40 entry points. 

41 """ 

42 try: 

43 eps = im.entry_points() 

44 return list(eps.select(group="bijux_cli.plugins")) 

45 except Exception: 

46 return [] 

47 

48 

49def _compatible(plugin: Any) -> bool: 

50 """Determines if a plugin is compatible with the current CLI API version. 

51 

52 Args: 

53 plugin (Any): The plugin instance or class, which should have a 

54 `requires_api_version` attribute string (e.g., ">=1.2.0"). 

55 

56 Returns: 

57 bool: True if the plugin's version requirement is met by the host CLI's 

58 API version, otherwise False. 

59 """ 

60 import bijux_cli 

61 

62 spec = getattr(plugin, "requires_api_version", ">=0.0.0") 

63 try: 

64 apiv = bijux_cli.api_version 

65 host_api_version = PkgVersion(str(apiv)) 

66 return SpecifierSet(spec).contains(host_api_version) 

67 except Exception: 

68 return False 

69 

70 

71async def load_entrypoints( 

72 di: DIContainer | None = None, 

73 registry: RegistryProtocol | None = None, 

74) -> None: 

75 """Discovers, loads, and registers all entry point-based plugins. 

76 

77 This function iterates through all entry points in the 'bijux_cli.plugins' 

78 group. For each one, it attempts to load, instantiate, and register the 

79 plugin. It also performs an API version compatibility check and runs the 

80 plugin's `startup` hook if present. 

81 

82 Note: 

83 All exceptions during the loading or startup of a single plugin are 

84 caught, logged, and reported via telemetry. A failed plugin will be 

85 deregistered and will not prevent other plugins from loading. 

86 

87 Args: 

88 di (DIContainer | None): The dependency injection container. If None, 

89 the current global container is used. 

90 registry (RegistryProtocol | None): The plugin registry. If None, it is 

91 resolved from the DI container. 

92 

93 Returns: 

94 None: 

95 """ 

96 import bijux_cli 

97 

98 di = di or DIContainer.current() 

99 registry = registry or di.resolve(RegistryProtocol) 

100 

101 obs = di.resolve(ObservabilityProtocol, None) 

102 tel = di.resolve(TelemetryProtocol, None) 

103 

104 for ep in _iter_plugin_eps(): 

105 try: 

106 plugin_class = ep.load() 

107 plugin = plugin_class() 

108 

109 if not _compatible(plugin): 

110 raise RuntimeError( 

111 f"Plugin '{ep.name}' requires API {getattr(plugin, 'requires_api_version', 'N/A')}, " 

112 f"host is {bijux_cli.api_version}" 

113 ) 

114 

115 for tgt in (plugin_class, plugin): 

116 raw = getattr(tgt, "version", None) 

117 if raw is not None and not isinstance(raw, str): 

118 tgt.version = str(raw) 

119 

120 registry.register(ep.name, plugin, version=plugin.version) 

121 

122 startup = getattr(plugin, "startup", None) 

123 if asyncio.iscoroutinefunction(startup): 

124 await startup(di) 

125 elif callable(startup): 

126 startup(di) 

127 

128 if obs: 

129 obs.log("info", f"Loaded plugin '{ep.name}'") 

130 if tel: 

131 tel.event("entrypoint_plugin_loaded", {"name": ep.name}) 

132 

133 except Exception as exc: 

134 with contextlib.suppress(Exception): 

135 registry.deregister(ep.name) 

136 

137 if obs: 

138 obs.log( 

139 "error", 

140 f"Failed to load plugin '{ep.name}'", 

141 extra={"trace": traceback.format_exc(limit=5)}, 

142 ) 

143 if tel: 

144 tel.event( 

145 "entrypoint_plugin_failed", {"name": ep.name, "error": str(exc)} 

146 ) 

147 

148 _LOG.debug("Skipped plugin %s: %s", ep.name, exc, exc_info=True) 

149 

150 

151__all__ = ["load_entrypoints"]