From cc1aa2338492d237a6a012501f50292ee77d81cb Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:19:27 -0700 Subject: [PATCH] Add lazy entry point extension discovery --- README.md | 10 + docs/extensions.md | 49 +++++ lib/python/base_cli/__init__.py | 26 ++- lib/python/base_cli/extensions.py | 323 ++++++++++++++++++++++++++++++ tests/test_extensions.py | 119 +++++++++++ 5 files changed, 526 insertions(+), 1 deletion(-) create mode 100644 docs/extensions.md create mode 100644 lib/python/base_cli/extensions.py create mode 100644 tests/test_extensions.py diff --git a/README.md b/README.md index 502460e..8f81f84 100644 --- a/README.md +++ b/README.md @@ -124,6 +124,16 @@ The core lifecycle is synchronous so cleanup, Click resource unwinding, and outcome finalization remain deterministic; an adapter may provide an explicit async runner without changing the core contract. +### Optional entry-point extensions + +Applications that want package-distributed commands, profiles, or integrations +can opt into `base_cli.ExtensionDiscovery`. It recognizes the documented +`base_cli.commands`, `base_cli.profiles`, and `base_cli.plugins` entry-point +groups. Discovery is lazy and cached, duplicate names fail explicitly, broken +extensions are isolated by `load_all()`, and consumers can disable discovery or +provide an allowlist. See [`docs/extensions.md`](docs/extensions.md) for the +entry-point contracts and deterministic ordering rules. + ## Public API The supported facade is `import base_cli`. It exports the command lifecycle diff --git a/docs/extensions.md b/docs/extensions.md new file mode 100644 index 0000000..f04f6aa --- /dev/null +++ b/docs/extensions.md @@ -0,0 +1,49 @@ +# Entry-point extensions + +`base_cli.ExtensionDiscovery` provides an optional, lazy discovery boundary for +framework extensions. The core package only reads package metadata when a +consumer creates a discovery instance; third-party code is imported only when +an extension is explicitly loaded. + +The supported entry-point groups are: + +| Group | Contract | Purpose | +| --- | --- | --- | +| `base_cli.commands` | A callable command registrar | Add commands to a consumer-owned `App` or Click tree | +| `base_cli.profiles` | A callable profile factory | Supply a `CliProfile` for a named consumer | +| `base_cli.plugins` | A callable plugin installer | Register a coordinated extension with a consumer | + +For example, a package can publish: + +```toml +[project.entry-points."base_cli.commands"] +audit = "acme_cli.audit:register" + +[project.entry-points."base_cli.profiles"] +acme = "acme_cli.profile:build_profile" + +[project.entry-points."base_cli.plugins"] +telemetry = "acme_cli.telemetry:install" +``` + +The loaded callable receives the arguments documented by the consuming +application. `base-cli` intentionally discovers metadata without imposing a +single command-tree or profile-construction shape; this keeps Click, Typer, +and consumer-owned composition boundaries independent. + +## Determinism and safety + +Descriptors are ordered by group, entry-point name, distribution, version, and +target value. Duplicate names are an error; installation order is never an +implicit precedence rule. Each descriptor retains distribution/version/extras +metadata for diagnostics and policy decisions. + +Discovery is cached per `ExtensionDiscovery` instance. Call `refresh()` after a +runtime environment change. `load_all()` isolates broken third-party imports +and returns an `ExtensionLoadResult` for every descriptor so one broken plugin +does not hide healthy extensions. + +Use `allowlist={"base_cli.commands:audit"}` to restrict names, or +`ExtensionDiscovery(disabled=True)` to disable discovery entirely. Allowlist +entries may be a bare entry-point name, a fully-qualified `group:name`, or a +distribution name. diff --git a/lib/python/base_cli/__init__.py b/lib/python/base_cli/__init__.py index ff054c6..0ea75cf 100644 --- a/lib/python/base_cli/__init__.py +++ b/lib/python/base_cli/__init__.py @@ -30,7 +30,7 @@ def _resolve_version() -> str: __version__ = _resolve_version() -from . import command_filters, command_protocol, history, json_contracts, testing +from . import command_filters, command_protocol, extensions, history, json_contracts, testing from .attachment import ( AttachmentAdapter, AttachmentContextFactory, @@ -72,6 +72,19 @@ def _resolve_version() -> str: get_current_context, ) from .errors import ConfigurationError +from .extensions import ( + COMMAND_ENTRY_POINT_GROUP, + ENTRY_POINT_GROUPS, + PLUGIN_ENTRY_POINT_GROUP, + PROFILE_ENTRY_POINT_GROUP, + ExtensionCollisionError, + ExtensionDescriptor, + ExtensionDiscovery, + ExtensionDiscoveryError, + ExtensionLoadError, + ExtensionLoadResult, + ExtensionsDisabledError, +) from .exit_codes import ExitCode from .inspection import inspection_envelope, render_inspection_json from .json_contracts import ( @@ -139,8 +152,17 @@ def _resolve_version() -> str: "CommandProtocolError", "CommandSchemaRegistry", "ConfigurationError", + "COMMAND_ENTRY_POINT_GROUP", "Context", "ConfigT", + "ENTRY_POINT_GROUPS", + "ExtensionCollisionError", + "ExtensionDescriptor", + "ExtensionDiscovery", + "ExtensionDiscoveryError", + "ExtensionLoadError", + "ExtensionLoadResult", + "ExtensionsDisabledError", "DEFAULT_SCHEMA_REGISTRY", "DisplayCommandResolver", "EnvironmentConfigLoader", @@ -195,6 +217,8 @@ def _resolve_version() -> str: "ProjectDiscovery", "RECORD_SCHEMAS", "RuntimeLayout", + "PLUGIN_ENTRY_POINT_GROUP", + "PROFILE_ENTRY_POINT_GROUP", "RetentionPolicy", "RuntimeResolver", "is_terminal", diff --git a/lib/python/base_cli/extensions.py b/lib/python/base_cli/extensions.py new file mode 100644 index 0000000..5b853f7 --- /dev/null +++ b/lib/python/base_cli/extensions.py @@ -0,0 +1,323 @@ +"""Lazy, deterministic discovery of optional Python entry-point extensions. + +The core package deliberately does not import or execute third-party plugins +at import time. Consumers opt into discovery with :class:`ExtensionDiscovery` +and choose when a selected entry point is loaded. +""" + +from __future__ import annotations + +import importlib.metadata as metadata +from collections.abc import Callable, Iterable, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from threading import RLock +from typing import Any, Protocol, cast + + +COMMAND_ENTRY_POINT_GROUP = "base_cli.commands" +PROFILE_ENTRY_POINT_GROUP = "base_cli.profiles" +PLUGIN_ENTRY_POINT_GROUP = "base_cli.plugins" +ENTRY_POINT_GROUPS = ( + COMMAND_ENTRY_POINT_GROUP, + PROFILE_ENTRY_POINT_GROUP, + PLUGIN_ENTRY_POINT_GROUP, +) + +__all__ = [ + "COMMAND_ENTRY_POINT_GROUP", + "ENTRY_POINT_GROUPS", + "ExtensionCollisionError", + "ExtensionDescriptor", + "ExtensionDiscovery", + "ExtensionDiscoveryError", + "ExtensionLoadError", + "ExtensionLoadResult", + "ExtensionsDisabledError", + "PLUGIN_ENTRY_POINT_GROUP", + "PROFILE_ENTRY_POINT_GROUP", +] + + +class ExtensionDiscoveryError(RuntimeError): + """Base class for actionable extension discovery failures.""" + + +class ExtensionsDisabledError(ExtensionDiscoveryError): + """Raised when a caller attempts to load an extension while disabled.""" + + +class ExtensionCollisionError(ExtensionDiscoveryError): + """Raised when more than one distribution claims the same extension name.""" + + def __init__(self, group: str, name: str, descriptors: Sequence["ExtensionDescriptor"]) -> None: + self.group = group + self.name = name + self.descriptors = tuple(descriptors) + details = ", ".join( + f"{descriptor.distribution or ''} {descriptor.version or ''}" + for descriptor in self.descriptors + ) + super().__init__( + f"Extension name '{name}' is claimed more than once in group '{group}': {details}. " + "Rename one entry point or select a different name explicitly." + ) + + +class ExtensionLoadError(ExtensionDiscoveryError): + """Wrap an extension import failure without hiding its source metadata.""" + + def __init__(self, descriptor: "ExtensionDescriptor", cause: BaseException) -> None: + self.descriptor = descriptor + self.cause = cause + super().__init__( + f"Unable to load {descriptor.group} extension '{descriptor.name}' " + f"from {descriptor.distribution or ''} " + f"({descriptor.value}): {cause}" + ) + + +@dataclass(frozen=True) +class ExtensionDescriptor: + """Stable metadata for one entry point, before its object is loaded.""" + + group: str + name: str + value: str + distribution: str | None + version: str | None + extras: tuple[str, ...] = () + + @property + def key(self) -> str: + """Return the fully qualified allowlist key ``group:name``.""" + + return f"{self.group}:{self.name}" + + +@dataclass(frozen=True) +class ExtensionLoadResult: + """Result of an isolated bulk extension load.""" + + descriptor: ExtensionDescriptor + value: Any | None = None + error: ExtensionLoadError | None = None + + @property + def ok(self) -> bool: + return self.error is None + + +class EntryPointProvider(Protocol): + """Provide an iterable of importlib metadata entry-point objects.""" + + def __call__(self) -> Iterable[Any]: ... + + +class ExtensionDiscovery: + """Discover and lazily load command, profile, and plugin entry points. + + Discovery is metadata-only until :meth:`load` is called. Results are + cached per instance; call :meth:`refresh` when the environment has changed. + Entry-point names are sorted deterministically by group, name, distribution, + version, and target value. Duplicate names are never resolved by + installation order: :meth:`load` raises :class:`ExtensionCollisionError`. + """ + + def __init__( + self, + *, + disabled: bool = False, + allowlist: Iterable[str] | None = None, + entry_points: Iterable[Any] | EntryPointProvider | None = None, + paths: Iterable[Path] | None = None, + ) -> None: + if entry_points is not None and paths is not None: + raise ValueError("pass either entry_points or paths, not both") + self.disabled = disabled + self.allowlist = frozenset(allowlist) if allowlist is not None else None + self._entry_points = entry_points + self._paths = tuple(Path(path) for path in paths) if paths is not None else None + self._raw_cache: tuple[Any, ...] | None = None + self._metadata_cache: tuple[ExtensionDescriptor, ...] | None = None + self._descriptor_cache: dict[str, tuple[ExtensionDescriptor, ...]] = {} + self._loaded_cache: dict[tuple[str, str], Any] = {} + self._lock = RLock() + + def list(self, group: str | None = None) -> tuple[ExtensionDescriptor, ...]: + """Return allowed metadata descriptors without importing extensions.""" + + if self.disabled: + return () + if group is not None: + _validate_group(group) + with self._lock: + if group is not None and group in self._descriptor_cache: + return self._descriptor_cache[group] + descriptors = self._metadata_descriptors() + if group is not None: + descriptors = tuple(descriptor for descriptor in descriptors if descriptor.group == group) + self._descriptor_cache[group] = descriptors + return descriptors + + def list_commands(self) -> tuple[ExtensionDescriptor, ...]: + """Return command extension descriptors.""" + + return self.list(COMMAND_ENTRY_POINT_GROUP) + + def list_profiles(self) -> tuple[ExtensionDescriptor, ...]: + """Return profile extension descriptors.""" + + return self.list(PROFILE_ENTRY_POINT_GROUP) + + def list_plugins(self) -> tuple[ExtensionDescriptor, ...]: + """Return plugin extension descriptors.""" + + return self.list(PLUGIN_ENTRY_POINT_GROUP) + + def load(self, group: str, name: str) -> Any: + """Load one uniquely named extension and cache the resulting object.""" + + if self.disabled: + raise ExtensionsDisabledError("Python extension discovery is disabled") + _validate_group(group) + matches = tuple(descriptor for descriptor in self.list(group) if descriptor.name == name) + if not matches: + raise ExtensionDiscoveryError(f"No extension named '{name}' exists in group '{group}'.") + if len(matches) > 1: + raise ExtensionCollisionError(group, name, matches) + key = (group, name) + with self._lock: + if key in self._loaded_cache: + return self._loaded_cache[key] + descriptor = matches[0] + try: + value = self._load_descriptor(descriptor) + except BaseException as exc: # isolate third-party import failures + raise ExtensionLoadError(descriptor, exc) from exc + with self._lock: + self._loaded_cache[key] = value + return value + + def load_all(self, group: str) -> tuple[ExtensionLoadResult, ...]: + """Load every allowed extension independently, preserving good results.""" + + if self.disabled: + raise ExtensionsDisabledError("Python extension discovery is disabled") + _validate_group(group) + results: list[ExtensionLoadResult] = [] + for descriptor in self.list(group): + try: + results.append(ExtensionLoadResult(descriptor, value=self.load(group, descriptor.name))) + except ExtensionLoadError as exc: + results.append(ExtensionLoadResult(descriptor, error=exc)) + except ExtensionCollisionError as exc: + results.append(ExtensionLoadResult(descriptor, error=ExtensionLoadError(descriptor, exc))) + return tuple(results) + + def refresh(self) -> None: + """Clear metadata and loaded-object caches for a new environment snapshot.""" + + with self._lock: + self._raw_cache = None + self._metadata_cache = None + self._descriptor_cache.clear() + self._loaded_cache.clear() + + def _metadata_descriptors(self) -> tuple[ExtensionDescriptor, ...]: + if self._metadata_cache is not None: + return self._metadata_cache + descriptors: list[ExtensionDescriptor] = [] + for entry_point in self._raw_entry_points(): + group = getattr(entry_point, "group", None) + name = getattr(entry_point, "name", None) + value = getattr(entry_point, "value", None) + if group not in ENTRY_POINT_GROUPS or not isinstance(name, str) or not isinstance(value, str): + continue + descriptor = _descriptor_from_entry_point(entry_point) + if self._allowed(descriptor): + descriptors.append(descriptor) + descriptors.sort(key=_descriptor_sort_key) + self._metadata_cache = tuple(descriptors) + return self._metadata_cache + + def _raw_entry_points(self) -> tuple[Any, ...]: + if self._raw_cache is not None: + return self._raw_cache + if self._paths is not None: + values: list[Any] = [] + for distribution in metadata.distributions(path=[str(path) for path in self._paths]): + values.extend(distribution.entry_points) + self._raw_cache = tuple(values) + return self._raw_cache + source = self._entry_points + if source is None: + self._raw_cache = _normalise_entry_points(metadata.entry_points()) + else: + raw_values: Any = source() if callable(source) else source + self._raw_cache = _normalise_entry_points(raw_values) + return self._raw_cache + + def _allowed(self, descriptor: ExtensionDescriptor) -> bool: + if self.allowlist is None: + return True + return bool( + descriptor.name in self.allowlist + or descriptor.key in self.allowlist + or (descriptor.distribution is not None and descriptor.distribution in self.allowlist) + ) + + def _load_descriptor(self, descriptor: ExtensionDescriptor) -> Any: + for entry_point in self._raw_entry_points(): + if ( + getattr(entry_point, "group", None) == descriptor.group + and getattr(entry_point, "name", None) == descriptor.name + and getattr(entry_point, "value", None) == descriptor.value + ): + return entry_point.load() + raise ImportError("entry point disappeared before it could be loaded") + + +def _validate_group(group: str) -> None: + if group not in ENTRY_POINT_GROUPS: + expected = ", ".join(ENTRY_POINT_GROUPS) + raise ValueError(f"Unsupported extension group '{group}'. Expected one of: {expected}.") + + +def _normalise_entry_points(values: Any) -> tuple[Any, ...]: + if isinstance(values, Mapping): + return tuple(entry_point for group in values.values() for entry_point in group) + return tuple(values) + + +def _descriptor_from_entry_point(entry_point: Any) -> ExtensionDescriptor: + distribution = getattr(entry_point, "dist", None) + distribution_name: str | None = None + version: str | None = None + if distribution is not None: + distribution_name = getattr(distribution, "name", None) + version = getattr(distribution, "version", None) + if distribution_name is None: + distribution_metadata = getattr(distribution, "metadata", None) + if distribution_metadata is not None: + distribution_name = distribution_metadata.get("Name") + extras = getattr(entry_point, "extras", ()) or () + return ExtensionDescriptor( + group=cast(str, entry_point.group), + name=cast(str, entry_point.name), + value=cast(str, entry_point.value), + distribution=distribution_name, + version=str(version) if version is not None else None, + extras=tuple(str(extra) for extra in extras), + ) + + +def _descriptor_sort_key(descriptor: ExtensionDescriptor) -> tuple[str, ...]: + return ( + descriptor.group.casefold(), + descriptor.name.casefold(), + descriptor.name, + (descriptor.distribution or "").casefold(), + descriptor.version or "", + descriptor.value, + ) diff --git a/tests/test_extensions.py b/tests/test_extensions.py new file mode 100644 index 0000000..4e143ab --- /dev/null +++ b/tests/test_extensions.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import importlib +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +import base_cli + + +def _entry_point( + name: str, + value: str, + *, + group: str = base_cli.COMMAND_ENTRY_POINT_GROUP, + distribution: str = "sample-package", + version: str = "1.0", +) -> SimpleNamespace: + return SimpleNamespace( + name=name, + value=value, + group=group, + extras=(), + dist=SimpleNamespace(name=distribution, version=version), + ) + + +class ExtensionDiscoveryTests(unittest.TestCase): + def test_metadata_discovery_is_lazy_deterministic_and_cached(self) -> None: + loaded = mock.Mock(return_value="command") + entry_point = _entry_point("audit", "sample:register") + entry_point.load = loaded + calls: list[None] = [] + + def provider() -> tuple[SimpleNamespace]: + calls.append(None) + return (entry_point,) + + discovery = base_cli.ExtensionDiscovery(entry_points=provider) + descriptors = discovery.list_commands() + self.assertEqual(descriptors[0].key, "base_cli.commands:audit") + self.assertEqual(descriptors[0].distribution, "sample-package") + self.assertEqual(calls, [None]) + self.assertEqual(discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "audit"), "command") + self.assertEqual(discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "audit"), "command") + self.assertEqual(calls, [None]) + loaded.assert_called_once_with() + + def test_duplicate_names_fail_instead_of_using_installation_order(self) -> None: + first = _entry_point("audit", "one:register", distribution="one", version="1") + second = _entry_point("audit", "two:register", distribution="two", version="2") + discovery = base_cli.ExtensionDiscovery(entry_points=(first, second)) + + with self.assertRaises(base_cli.ExtensionCollisionError) as raised: + discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "audit") + self.assertIn("one", str(raised.exception)) + self.assertIn("two", str(raised.exception)) + + def test_allowlist_and_disable_switch_are_enforced(self) -> None: + allowed = _entry_point("allowed", "one:register") + blocked = _entry_point("blocked", "two:register") + discovery = base_cli.ExtensionDiscovery( + entry_points=(allowed, blocked), + allowlist={"base_cli.commands:allowed"}, + ) + self.assertEqual([item.name for item in discovery.list_commands()], ["allowed"]) + + disabled = base_cli.ExtensionDiscovery(entry_points=(allowed,), disabled=True) + self.assertEqual(disabled.list_commands(), ()) + with self.assertRaises(base_cli.ExtensionsDisabledError): + disabled.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "allowed") + + def test_load_all_isolates_broken_extensions(self) -> None: + healthy = _entry_point("healthy", "one:register") + broken = _entry_point("broken", "two:register") + healthy.load = lambda: "healthy" + broken.load = lambda: (_ for _ in ()).throw(ImportError("missing optional dependency")) + results = base_cli.ExtensionDiscovery(entry_points=(healthy, broken)).load_all( + base_cli.COMMAND_ENTRY_POINT_GROUP + ) + + self.assertEqual([result.descriptor.name for result in results], ["broken", "healthy"]) + self.assertTrue(results[1].ok) + self.assertFalse(results[0].ok) + self.assertIn("missing optional dependency", str(results[0].error)) + + def test_real_distribution_metadata_is_discovered_from_a_path(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + (root / "sample_extension.py").write_text( + "def register():\n return 'installed'\n", + encoding="utf-8", + ) + dist_info = root / "sample_extension-1.2.3.dist-info" + dist_info.mkdir() + (dist_info / "METADATA").write_text( + "Metadata-Version: 2.1\nName: sample-extension\nVersion: 1.2.3\n", + encoding="utf-8", + ) + (dist_info / "entry_points.txt").write_text( + "[base_cli.commands]\naudit = sample_extension:register\n", + encoding="utf-8", + ) + sys.path.insert(0, str(root)) + try: + importlib.invalidate_caches() + discovery = base_cli.ExtensionDiscovery(paths=(root,)) + descriptors = discovery.list_commands() + self.assertEqual(descriptors[0].version, "1.2.3") + self.assertEqual(discovery.load(base_cli.COMMAND_ENTRY_POINT_GROUP, "audit")(), "installed") + finally: + sys.path.remove(str(root)) + + +if __name__ == "__main__": + unittest.main()