Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / plugins / loader.py: 90%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Lazy plugin command loading helpers.""" 

5 

6from __future__ import annotations 

7 

8import importlib.util 

9import sys 

10from types import ModuleType 

11from typing import Any 

12 

13import click 

14import typer 

15 

16from bijux_cli.core.runtime import AsyncTyper, adapt_typer 

17from bijux_cli.plugins.metadata import ( 

18 PluginMetadata, 

19 PluginMetadataError, 

20 validate_plugin_metadata, 

21) 

22 

23 

24def _load_module_from_path(path: str, module_name: str) -> ModuleType: 

25 """Load a module from a file path and register it in sys.modules.""" 

26 spec = importlib.util.spec_from_file_location(module_name, path) 

27 if not spec or not spec.loader: 

28 raise PluginMetadataError(f"Cannot import plugin: {path}", http_status=400) 

29 module = importlib.util.module_from_spec(spec) 

30 sys.modules[module_name] = module 

31 try: 

32 spec.loader.exec_module(module) 

33 except (FileNotFoundError, OSError) as exc: 

34 raise PluginMetadataError( 

35 f"Cannot import plugin: {path}", http_status=400 

36 ) from exc 

37 return module 

38 

39 

40def _load_typer_from_module(module: ModuleType) -> typer.Typer: 

41 """Extract a Typer app from a plugin module and adapt it for async.""" 

42 if hasattr(module, "cli") and callable(module.cli): 

43 app = module.cli() 

44 elif hasattr(module, "app"): 

45 app = module.app 

46 else: 

47 raise PluginMetadataError( 

48 "Plugin has no CLI entrypoint (expected cli() or app)", 

49 http_status=400, 

50 ) 

51 if not isinstance(app, typer.Typer): 

52 raise PluginMetadataError( 

53 "Plugin CLI entrypoint did not return a Typer app", 

54 http_status=400, 

55 ) 

56 adapt_typer(app) 

57 return app 

58 

59 

60def _entrypoint_loader(meta: PluginMetadata) -> typer.Typer: 

61 """Load a Typer app from entry point metadata.""" 

62 if not meta.entrypoint: 

63 raise PluginMetadataError("Entry point metadata missing", http_status=400) 

64 obj = meta.entrypoint.load() 

65 if isinstance(obj, typer.Typer): 

66 adapt_typer(obj) 

67 return obj 

68 if callable(obj): 

69 obj = obj() 

70 if hasattr(obj, "registered_groups"): 

71 app = typer.Typer() 

72 for name, sub in getattr(obj, "registered_groups", {}).items(): 

73 if isinstance(sub, typer.Typer): 73 ↛ 72line 73 didn't jump to line 72 because the condition on line 73 was always true

74 adapt_typer(sub) 

75 app.add_typer(sub, name=name) 

76 return app 

77 if hasattr(obj, "register") and callable(obj.register): 

78 app = typer.Typer() 

79 obj.register(app) 

80 adapt_typer(app) 

81 return app 

82 if hasattr(obj, "app") and isinstance(obj.app, typer.Typer): 

83 adapt_typer(obj.app) 

84 return obj.app 

85 raise PluginMetadataError( 

86 f"Entry point {meta.name!r} did not provide a Typer app", 

87 http_status=400, 

88 ) 

89 

90 

91def _local_loader(meta: PluginMetadata) -> typer.Typer: 

92 """Load a local plugin by importing its plugin.py module.""" 

93 if not meta.path: 

94 raise PluginMetadataError("Local plugin path missing", http_status=400) 

95 plug_py = meta.path / "plugin.py" 

96 module = _load_module_from_path(str(plug_py), f"_bijux_cli_plugin_{meta.name}") 

97 return _load_typer_from_module(module) 

98 

99 

100class LazyTyper(AsyncTyper): 

101 """Typer app that loads a plugin Typer app on first access.""" 

102 

103 def __init__(self, meta: PluginMetadata): 

104 """Initialize a lazy-loading Typer wrapper.""" 

105 super().__init__(name=meta.name, invoke_without_command=True) 

106 self._meta = meta 

107 self._loaded: click.Group | None = None 

108 

109 def _load(self) -> click.Group: 

110 """Load the underlying plugin command tree on first access.""" 

111 if self._loaded is None: 111 ↛ 123line 111 didn't jump to line 123 because the condition on line 111 was always true

112 if self._meta.source == "entrypoint": 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true

113 app = _entrypoint_loader(self._meta) 

114 else: 

115 app = _local_loader(self._meta) 

116 loaded = typer.main.get_command(app) 

117 if isinstance(loaded, click.Group): 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true

118 self._loaded = loaded 

119 else: 

120 wrapper = click.Group(name=self._meta.name) 

121 wrapper.add_command(loaded, loaded.name) 

122 self._loaded = wrapper 

123 return self._loaded 

124 

125 def list_commands(self, ctx: click.Context) -> list[str]: 

126 """List commands after loading the plugin.""" 

127 return self._load().list_commands(ctx) 

128 

129 def get_command(self, ctx: click.Context, name: str) -> click.Command | None: 

130 """Resolve a command after loading the plugin.""" 

131 return self._load().get_command(ctx, name) 

132 

133 def invoke(self, ctx: click.Context) -> Any: 

134 """Invoke the loaded plugin command group.""" 

135 return self._load().invoke(ctx) 

136 

137 

138def lazy_command_for(meta: PluginMetadata) -> typer.Typer: 

139 """Return a lazy-loading Typer wrapper for a plugin.""" 

140 return load_command_for(meta) 

141 

142 

143def load_command_for(meta: PluginMetadata) -> typer.Typer: 

144 """Load a plugin Typer app immediately.""" 

145 # Invariant: metadata validation happens before any activation attempt. 

146 validate_plugin_metadata(meta) 

147 if meta.source == "entrypoint": 147 ↛ 149line 147 didn't jump to line 149 because the condition on line 147 was always true

148 return _entrypoint_loader(meta) 

149 return _local_loader(meta) 

150 

151 

152def activate_plugin(meta: PluginMetadata) -> typer.Typer: 

153 """Activate a plugin and return its Typer command tree.""" 

154 return load_command_for(meta) 

155 

156 

157def deactivate_plugin(_meta: PluginMetadata) -> None: 

158 """Deactivate a plugin if applicable (no-op for now).""" 

159 return None