Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/services/plugins/__init__.py: 100%
150 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-19 23:36 +0000
1# SPDX-License-Identifier: MIT
2# Copyright © 2025 Bijan Mousavi
4"""Provides the public API for the Bijux CLI's plugin management service.
6This module acts as the public facade for the plugin management service layer.
7It exposes a curated set of high-level functions for core plugin operations,
8ensuring a stable and convenient API for the rest of the application.
10To optimize CLI startup performance, this module's own submodules (such as
11`registry`, `hooks`, and `entrypoints`) are loaded lazily upon first access.
12"""
14from __future__ import annotations
16from contextlib import suppress
17import importlib
18import importlib.util
19import os
20from pathlib import Path
21import shutil
22import sys
23from typing import Any, cast
25from packaging.specifiers import SpecifierSet
27from bijux_cli.__version__ import version as cli_version
28from bijux_cli.contracts import (
29 ObservabilityProtocol,
30 RegistryProtocol,
31 TelemetryProtocol,
32)
33from bijux_cli.core.exceptions import BijuxError
36def _di() -> Any | None:
37 """Safely retrieves the current DIContainer instance.
39 Returns:
40 DIContainer | None: The current dependency injection container, or None
41 if it is not available or an error occurs.
42 """
43 from bijux_cli.core.di import DIContainer
45 try:
46 return DIContainer.current()
47 except Exception:
48 return None
51def _obs() -> ObservabilityProtocol | None:
52 """Safely resolves the `ObservabilityProtocol` service.
54 Returns:
55 ObservabilityProtocol | None: The observability service, or None if it
56 cannot be resolved.
57 """
58 di = _di()
59 if not di:
60 return None
61 try:
62 return cast(ObservabilityProtocol, di.resolve(ObservabilityProtocol))
63 except KeyError:
64 return None
67def _tel() -> TelemetryProtocol | None:
68 """Safely resolves the `TelemetryProtocol` service.
70 Returns:
71 TelemetryProtocol | None: The telemetry service, or None if it
72 cannot be resolved.
73 """
74 di = _di()
75 if not di:
76 return None
77 try:
78 return cast(TelemetryProtocol, di.resolve(TelemetryProtocol))
79 except KeyError:
80 return None
83def get_plugins_dir() -> Path:
84 """Returns the directory that stores installed plugins.
86 The path is determined by the `BIJUXCLI_PLUGINS_DIR` environment variable
87 if set, otherwise it falls back to a default location. The directory is
88 created if it does not exist.
90 Returns:
91 Path: The resolved path to the plugins directory.
92 """
93 env_path = os.environ.get("BIJUXCLI_PLUGINS_DIR")
94 if env_path:
95 plugins_dir = Path(env_path)
96 else:
97 from bijux_cli.core.paths import PLUGINS_DIR
99 plugins_dir = PLUGINS_DIR
100 plugins_dir = plugins_dir.expanduser()
102 if plugins_dir.exists() and plugins_dir.is_symlink():
103 return plugins_dir
104 if plugins_dir.is_dir():
105 return plugins_dir.resolve()
106 if plugins_dir.exists():
107 return plugins_dir.resolve()
108 plugins_dir.mkdir(parents=True, exist_ok=True)
109 return plugins_dir.resolve()
112def load_plugin_config(name: str) -> dict[str, Any]:
113 """Loads a plugin's `config.yaml` file.
115 This function looks for a `config.yaml` file in the specified plugin's
116 directory. It returns an empty dictionary if the file is missing.
118 Args:
119 name (str): The name of the plugin whose config should be loaded.
121 Returns:
122 dict[str, Any]: The plugin's configuration as a dictionary.
124 Raises:
125 BijuxError: If the `PyYAML` library is not installed or if the file
126 is corrupt and cannot be parsed.
127 """
128 try:
129 import yaml
130 except ModuleNotFoundError as exc:
131 raise BijuxError("PyYAML is required to read plugin configs") from exc
133 cfg_path = get_plugins_dir() / name / "config.yaml"
134 if not cfg_path.exists():
135 return {}
137 try:
138 data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) or {}
139 except Exception as exc:
140 tel = _tel()
141 if tel:
142 tel.event("plugin_config_failed", {"name": name, "error": str(exc)})
143 raise BijuxError(f"Failed to load config for '{name}': {exc}") from exc
145 obs = _obs()
146 tel = _tel()
147 if obs:
148 obs.log("info", "Loaded plugin config", extra={"name": name})
149 if tel:
150 tel.event("plugin_config_loaded", {"name": name})
151 return data
154def verify_plugin_signature(path: Path, public_key: str | None) -> bool:
155 """Verifies the signature of a plugin file.
157 Note:
158 This is a placeholder function. Actual cryptographic verification is
159 not implemented.
161 Args:
162 path (Path): The path to the plugin file to verify.
163 public_key (str | None): The public key to use for verification.
165 Returns:
166 bool: True if the signature is "verified", False if no signature exists.
168 Raises:
169 BijuxError: If a signature file exists but no public key is provided.
170 """
171 sig = path.with_suffix(path.suffix + ".sig")
172 tel = _tel()
174 if not sig.exists():
175 if tel:
176 tel.event("plugin_unsigned", {"path": str(path)})
177 return False
179 if public_key is None:
180 raise BijuxError(f"Signature found for {path} but no public_key provided")
182 if tel:
183 tel.event("plugin_signature_verified", {"path": str(path)})
184 return True
187def load_plugin(
188 path: str | Path,
189 module_name: str,
190 *,
191 public_key: str | None = None,
192) -> Any:
193 """Dynamically loads and instantiates a `Plugin` class from a `.py` file.
195 This function handles the dynamic import of a plugin's code, instantiates
196 its main `Plugin` class, and performs compatibility checks.
198 Args:
199 path (str | Path): The path to the `plugin.py` file.
200 module_name (str): The name to assign to the imported module.
201 public_key (str | None): An optional public key for signature verification.
203 Returns:
204 Any: An instantiated object of the `Plugin` class from the file.
206 Raises:
207 BijuxError: If the file is missing, the signature is invalid, the
208 plugin is incompatible with the current CLI version, or an import
209 error occurs.
210 """
211 path = Path(path)
212 if not path.is_file():
213 raise BijuxError(f"Plugin file not found: {path}")
215 if public_key:
216 verify_plugin_signature(path, public_key)
218 spec = importlib.util.spec_from_file_location(module_name, str(path))
219 if not spec or not spec.loader:
220 raise BijuxError(f"Cannot import plugin: {path}")
222 module = importlib.util.module_from_spec(spec)
223 sys.modules[module_name] = module
225 try:
226 spec.loader.exec_module(module)
228 plugin_class = getattr(module, "Plugin", None)
229 if plugin_class is None:
230 raise BijuxError("No `Plugin` class found in module")
232 plugin = plugin_class()
234 for target in (plugin_class, plugin):
235 raw = getattr(target, "version", None)
236 if raw is not None and not isinstance(raw, str):
237 target.version = str(raw)
239 required = getattr(plugin, "requires_cli_version", f"=={cli_version}")
240 spec_set = SpecifierSet(required)
241 if not spec_set.contains(cli_version):
242 raise BijuxError(f"Plugin requires CLI {required}, host is {cli_version}")
244 obs = _obs()
245 cli_attr = getattr(plugin, "cli", None)
246 if obs and (cli_attr is None or not callable(cli_attr)):
247 obs.log(
248 "warning",
249 f"Plugin '{module_name}' has no callable `cli`",
250 extra={"path": str(path)},
251 )
253 tel = _tel()
254 if tel:
255 tel.event("plugin_loaded", {"name": getattr(plugin, "name", module_name)})
257 return plugin
259 except Exception as exc:
260 sys.modules.pop(module_name, None)
261 raise BijuxError(f"Failed to load plugin '{path}': {exc}") from exc
264def uninstall_plugin(name: str, registry: RegistryProtocol) -> bool:
265 """Removes a plugin's directory and deregisters it.
267 Args:
268 name (str): The name of the plugin to uninstall.
269 registry (RegistryProtocol): The plugin registry service.
271 Returns:
272 bool: True if the plugin was found and removed, otherwise False.
273 """
274 plug_dir = get_plugins_dir() / name
275 tel = _tel()
277 existed = registry.has(name)
278 if not existed:
279 if tel:
280 tel.event("plugin_uninstall_not_found", {"name": name})
281 return False
283 with suppress(Exception):
284 shutil.rmtree(plug_dir, ignore_errors=True)
286 registry.deregister(name)
287 if tel:
288 tel.event("plugin_uninstalled", {"name": name})
289 return True
292def install_plugin(*args: Any, **kwargs: Any) -> None:
293 """Stub for plugin installation. Use the CLI command instead.
295 Raises:
296 NotImplementedError: Always, as this function is a stub.
297 """
298 raise NotImplementedError("Use `bijux_cli.commands.plugins.install`")
301_SUBMODULES: dict[str, str] = {
302 "hooks": "bijux_cli.services.plugins.hooks",
303 "entrypoints": "bijux_cli.services.plugins.entrypoints",
304 "groups": "bijux_cli.services.plugins.groups",
305 "registry": "bijux_cli.services.plugins.registry",
306}
309def __getattr__(name: str) -> Any:
310 """Lazily imports submodules to optimize startup time.
312 This function is a module-level implementation of `__getattr__` (PEP 562),
313 which allows submodules of the `plugins` service to be imported only when
314 they are first accessed.
316 Args:
317 name (str): The name of the submodule or attribute to access.
319 Returns:
320 Any: The imported submodule or attribute.
322 Raises:
323 AttributeError: If the requested name is not a valid submodule or
324 attribute that can be lazily loaded.
325 """
326 if name in _SUBMODULES:
327 mod = importlib.import_module(_SUBMODULES[name])
328 setattr(sys.modules[__name__], name, mod)
329 return mod
330 if name in {"command_group", "dynamic_choices"}:
331 groups_mod = importlib.import_module(_SUBMODULES["groups"])
332 return getattr(groups_mod, name)
333 if name == "load_entrypoints":
334 entrypoints_mod = importlib.import_module(_SUBMODULES["entrypoints"])
335 return entrypoints_mod.load_entrypoints
336 raise AttributeError(f"module {__name__} has no attribute {name}")
339__all__ = [
340 "get_plugins_dir",
341 "load_plugin_config",
342 "verify_plugin_signature",
343 "load_plugin",
344 "uninstall_plugin",
345 "install_plugin",
346]