mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-08-28 05:25:04 +00:00
Some checks are pending
CI / Documentation / Test and build documentation (push) Waiting to run
CI / Python quality / Pre-commit (push) Waiting to run
CI / Python tests / Unit Tests - py3.11 (push) Waiting to run
CI / Python tests / Unit Tests - py3.12 (push) Waiting to run
CI / Python tests / Unit Tests - py3.13 (push) Waiting to run
CI / Windows / CLI smoke - py3.11 (push) Waiting to run
Deploy / Documentation / Build documentation (push) Waiting to run
Deploy / Documentation / deploy (push) Blocked by required conditions
Security / CodeQL / Analyze javascript-typescript (push) Waiting to run
Security / CodeQL / Analyze python (push) Waiting to run
* feat: simplify plugin setup and add management CLI * fix: isolate plugin CLI import side effects * refactor: streamline plugin validation * fix: route plugin CLI arguments independently * fix: support standard plugin source layouts
35 lines
1.3 KiB
Python
35 lines
1.3 KiB
Python
"""Safe loading boundary for third-party Python entry points."""
|
|
|
|
from importlib import metadata
|
|
from typing import Any
|
|
|
|
PLUGIN_ENTRY_POINT_GROUP = "reme.plugins"
|
|
CONFIG_ENTRY_POINT_GROUP = "reme.configs"
|
|
|
|
|
|
def find_entry_points(group: str, name: str) -> list[metadata.EntryPoint]:
|
|
"""Return all matching entry points without importing their providers."""
|
|
return list(metadata.entry_points().select(group=group, name=name))
|
|
|
|
|
|
def unique_entry_point(
|
|
entries: list[metadata.EntryPoint],
|
|
name: str,
|
|
*,
|
|
provider: str,
|
|
) -> metadata.EntryPoint | None:
|
|
"""Return the sole matching entry point, rejecting ambiguous providers."""
|
|
if len(entries) > 1:
|
|
values = ", ".join(sorted(entry.value for entry in entries))
|
|
raise ValueError(f"{provider} '{name}' has multiple installed providers: {values}")
|
|
return entries[0] if entries else None
|
|
|
|
|
|
def load_entry_point(entry: metadata.EntryPoint, *, invoke: bool = False) -> Any:
|
|
"""Load and optionally invoke an entry point without retaining registrations."""
|
|
# Import lazily so config parser imports do not pull in the component graph.
|
|
from .components.component_registry import R
|
|
|
|
with R.preserve(allow_mutation=True):
|
|
loaded = entry.load()
|
|
return loaded() if invoke and callable(loaded) else loaded
|