Registry¶

def register_plugins(self, plugins: dict[str, Any]) -> None:
    """Register pre-discovered plugins without scanning entry points.

    Args:
        plugins: Mapping of plugin name to plugin.
    """
    for name, plugin_class in plugins.items():
        label = self._label_for(plugin_class, name)
        if label is not None:
            self._registries[label].register(
                name, plugin_class, self._original_collections
            )

def discover_plugins(self) -> None:
    """Finds all installed packages declaring aereo entry_points.

    Returns:
        None. Populates internal registries as a side effect.
    """
    logger.info("Discovering aereo plugins...")

    # Load all plugins from the unified aereo.plugins group
    eps = importlib.metadata.entry_points(group=self.ENTRY_POINT_GROUP)
    for ep in eps:
        try:
            plugin_class = ep.load()
            label = self._label_for(plugin_class, ep.name)

            if label is not None:
                self._registries[label].register(
                    ep.name, plugin_class, self._original_collections
                )
                logger.debug(f"Loaded {label}: {ep.name}")
            else:
                logger.warning(
                    f"Plugin '{ep.name}' does not inherit from any known base class or signature. Skipping."
                )
        except Exception as e:
            logger.error(
                f"Failed to load plugin '{ep.name}': {e}",
                exc_info=True,
            )

def find_for(self, type_label: str, collection_name: str) -> list[str]:
    """Return plugin names of *type_label* supporting *collection_name*.

    Args:
        type_label: Plugin type label (e.g. "searcher", "reader", "writer").
        collection_name: Name of the collection to query.

    Returns:
        Sorted list of plugin names that support the collection.
    """
    registry = self._registries.get(type_label)
    if registry is None:
        return []
    return registry.find_for(collection_name, self.WILDCARD)

def get(self, type_label: str, plugin_name: str, **kwargs: Any) -> Any:
    """Instantiate a plugin of *type_label* by *plugin_name*.

    Args:
        type_label: Plugin type label (e.g. "searcher", "reader").
        plugin_name: Entry-point name of the plugin.
        **kwargs: Configuration values passed to the plugin constructor.

    Returns:
        An instantiated plugin.

    Raises:
        ValueError: If the plugin type or name is not known.
    """
    registry = self._registries.get(type_label)
    if registry is None:
        raise ValueError(f"Unknown plugin type: {type_label}")
    if not registry.has(plugin_name):
        self._try_lazy_load(plugin_name, type_label)
    return registry.get(plugin_name, type_label.capitalize(), **kwargs)

def has(self, type_label: str, plugin_name: str) -> bool:
    """Check whether a plugin of *type_label* with *plugin_name* is registered.

    Args:
        type_label: Plugin type label.
        plugin_name: Entry-point name of the plugin.

    Returns:
        True if registered (or successfully lazy-loaded), False otherwise.
    """
    registry = self._registries.get(type_label)
    if registry is None:
        return False
    if registry.has(plugin_name):
        return True
    return self._try_lazy_load(plugin_name, type_label)

def list_supported_collections(self) -> list[str]:
    """Returns a list of all products supported by currently installed plugins.

    Returns:
        Sorted list of collection names in their original case as defined
        by plugins.
    """
    all_products: set[str] = set()
    for registry in self._registries.values():
        all_products.update(registry.collection_to_plugins.keys())
    display_names = [self._original_collections.get(p, p) for p in all_products]
    return sorted(display_names)

def find_searchers_for(self, collection_name: str) -> list[str]:
    """Returns names of search plugins that support the requested collection."""
    return self._registries["searcher"].find_for(collection_name, self.WILDCARD)

def get_searcher_collections(self, plugin_name: str) -> list[str]:
    """Return the supported collections for a named search plugin."""
    return self._registries["searcher"].get_collections(plugin_name)

def has_searcher(self, plugin_name: str) -> bool:
    """Check whether a search plugin with the given name is registered."""
    return self.has("searcher", plugin_name)

def get_collection_mapping_for_searcher(
    self, plugin_name: str, collection_names: Sequence[str]
) -> list[str]:
    """Maps user-provided collection names to a specific search plugin's declared format."""
    return self._registries["searcher"].get_collection_mapping(
        plugin_name, collection_names, self.WILDCARD
    )

def get_searcher(self, plugin_name: str, **kwargs) -> SearchProvider:
    """Instantiates and returns a SearchProvider by name."""
    return self.get("searcher", plugin_name, **kwargs)

def get_plugin_params(
    self, plugin_name: str, *, detailed: bool = True
) -> dict[str, list[dict]]:
    """Return params metadata for any plugin (search or pipeline stage).

    Args:
        plugin_name: Entry-point name of the plugin.
        detailed: When ``True`` (default), each param includes all
            attributes.

    Returns:
        {"required": [...], "optional": [...]} representing parameters.
    """
    cls: Any | None = None
    for registry in self._registries.values():
        cls = registry.plugins.get(plugin_name)
        if cls is not None:
            break

    if cls is None:
        raise KeyError(f"Unknown plugin: {plugin_name}")

    return _dump_params_function(cls, detailed)

def list_all_params(self, *, detailed: bool = True) -> dict[str, dict]:
    """JSON-serializable params catalog for all discovered plugins.

    Args:
        detailed: When ``True`` (default), each param includes all
            attributes.

    Returns:
        Mapping of plugin name to a dict with keys ``type``, ``required``,
        and ``optional``.
    """
    result: dict[str, dict] = {}
    for label, registry in self._registries.items():
        for name, cls in registry.plugins.items():
            params_dict = _dump_params_function(cls, detailed)
            result[name] = {
                "type": label,
                "required": params_dict["required"],
                "optional": params_dict["optional"],
            }
    return result