Abstract away plugin lookup and querying

variantlib/commands/generate_index_json.py

@@ -5,44 +5,14 @@
 import logging
 import pathlib
 import zipfile
-from importlib.metadata import entry_points
-from typing import Iterable
 
 from variantlib.meta import VariantMeta
+from variantlib.plugins import PluginLoader
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
 
 
-def get_variant_plugins() -> Iterable[tuple[str, str]]:
-    logger.info("Discovering Wheel Variant plugins...")
-    plugins = entry_points().select(group="variantlib.plugins")
-
-    seen = set()
-    duplicates = set()
-    for plugin_name in [plugin.name for plugin in plugins]:
-        if plugin_name in seen:
-            duplicates.add(plugin_name)
-        else:
-            seen.add(plugin_name)
-
-    if duplicates:
-        logger.warning(
-            "Duplicate plugins found: %s - Unpredicatable behavior.", duplicates
-        )
-
-    for plugin in plugins:
-        try:
-            logger.info(
-                f"Loading plugin: {plugin.name} - v{plugin.dist.version}"
-            )  # noqa: G004
-            plugin_class = plugin.load()
-            # TODO: is __provider_name__ a public API?
-            yield (plugin_class.__provider_name__, plugin.dist.name)
-        except Exception:
-            logging.exception("An unknown error happened - Ignoring plugin")
-
-
 def generate_index_json(args):
     parser = argparse.ArgumentParser(
         prog="generate_index_json",
@@ -105,7 +75,7 @@ def generate_index_json(args):
                     f"{wheel}: different metadata assigned to {variant_hash}"
                 )
 
-    all_plugins = dict(get_variant_plugins())
+    all_plugins = PluginLoader().get_dist_name_mapping()
     provider_requires = set()
     for provider in known_providers:
         if (plugin := all_plugins.get(provider)) is not None:

variantlib/metaclasses.py

@@ -0,0 +1,16 @@
+#! /usr/bin/python
+# -*- coding: utf-8 -*-
+
+__all__ = [
+    "SingletonMetaClass",
+]
+
+
+class SingletonMetaClass(type):
+    _instances = {}
+
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super().__call__(*args, **kwargs)
+
+        return cls._instances[cls]

variantlib/platform.py

@@ -2,16 +2,16 @@
 
 import logging
 from collections import defaultdict
-from importlib.metadata import entry_points
 from typing import TYPE_CHECKING
 
 from variantlib.combination import filtered_sorted_variants
 from variantlib.combination import get_combinations
-from variantlib.config import ProviderConfig
+from variantlib.plugins import PluginLoader
 
 if TYPE_CHECKING:
     from collections.abc import Generator
 
+    from variantlib.config import ProviderConfig
     from variantlib.meta import VariantDescription
 
 logger = logging.getLogger(__name__)
@@ -34,59 +34,14 @@ def wrapper(*args, **kwargs):
 
 @VariantCache()
 def _query_variant_plugins() -> dict[str, ProviderConfig]:
-    logger.info("Discovering Wheel Variant plugins...")
-    plugins = entry_points().select(group="variantlib.plugins")
-
-    # ----------- Checking if two plugins have the same name ----------- #
-    seen = set()
-    duplicates = set()
-
-    for plugin_name in [plugin.name for plugin in plugins]:
-        if plugin_name in seen:
-            duplicates.add(plugin_name)
-        else:
-            seen.add(plugin_name)
-
-    if duplicates:
-        logger.warning(
-            "Duplicate plugins found: %s - Unpredicatable behavior.", duplicates
-        )
-
-    # ---------------------- Querying each plugin ---------------------- #
-    provider_cfgs = {}
-    for plugin in plugins:
-        try:
-            logger.info(f"Loading plugin: {plugin.name} - v{plugin.dist.version}")  # noqa: G004
-
-            # Dynamically load the plugin class
-            plugin_class = plugin.load()
-
-            # Instantiate the plugin
-            plugin_instance = plugin_class()
-
-            # Call the `run` method of the plugin
-            provider_cfg = plugin_instance.run()
-
-            if not isinstance(provider_cfg, ProviderConfig):
-                logging.error(
-                    f"Provider: {plugin.name} returned an unexpected type: "  # noqa: G004
-                    f"{type(provider_cfg)} - Expected: `ProviderConfig`. Ignoring..."
-                )
-                continue
-
-            provider_cfgs[plugin.name] = provider_cfg
-
-        except Exception:
-            logging.exception("An unknown error happened - Ignoring plugin")
-
-    return provider_cfgs
+    return PluginLoader().get_provider_configs()
 
 
 def get_variant_hashes_by_priority(
     provider_priority_dict: dict[str:int] | None = None,
     variants_json: dict | None = None,
 ) -> Generator[VariantDescription]:
-    plugins = entry_points().select(group="variantlib.plugins")
+    provider_cfgs = _query_variant_plugins()
 
     # sorting providers in priority order:
     if provider_priority_dict is not None:
@@ -118,23 +73,28 @@ def get_variant_hashes_by_priority(
                     logger.warning("Value: %s -> Keys: %s", value, keys)
 
             # ----------- Checking if two plugins hold the same priority ----------- #
-            for plugin in plugins:
-                if plugin.name not in provider_priority_dict:
+            for plugin_name in provider_cfgs:
+                if plugin_name not in provider_priority_dict:
                     logger.warning(
                         "Plugin: %s is not present in the `provider_priority_dict`. "
                         "Will be treated as lowest priority.",
-                        plugin.name,
+                        plugin_name,
                     )
                     continue
 
             # ------------------- Sorting the plugins by priority ------------------ #
             plugins = sorted(
-                plugins,
-                key=lambda plg: provider_priority_dict.get(plg.name, float("inf")),
+                provider_cfgs,
+                key=lambda plugin_name: provider_priority_dict.get(
+                    plugin_name, float("inf")
+                ),
             )
 
-    provider_cfgs = _query_variant_plugins()
-    sorted_provider_cfgs = [provider_cfgs[plugin.name] for plugin in plugins]
+            sorted_provider_cfgs = [
+                provider_cfgs[plugin_name] for plugin_name in plugins
+            ]
+    else:
+        sorted_provider_cfgs = list(provider_cfgs.values())
 
     if sorted_provider_cfgs:
         if (variants_json or {}).get("variants") is not None:

variantlib/plugins.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import logging
+from functools import cache
+from importlib.metadata import entry_points
+
+from variantlib.config import ProviderConfig
+from variantlib.metaclasses import SingletonMetaClass
+
+logger = logging.getLogger(__name__)
+
+
+class PluginLoader(metaclass=SingletonMeta):
+    """Load and query plugins"""
+
+    def __init__(self) -> None:
+        self._plugins = {}
+        self._dist_names = {}
+
+        load_plugins()
+
+    def load_plugins(self) -> None:
+        """Find, load and instantiate all plugins"""
+
+        logger.info("Discovering Wheel Variant plugins...")
+        plugins = entry_points().select(group="variantlib.plugins")
+
+        # ----------- Checking if two plugins have the same name ----------- #
+        seen = set()
+        duplicates = set()
+
+        for plugin_name in [plugin.name for plugin in plugins]:
+            if plugin_name in seen:
+                duplicates.add(plugin_name)
+            else:
+                seen.add(plugin_name)
+
+        if duplicates:
+            logger.warning(
+                "Duplicate plugins found: %s - Unpredicatable behavior.", duplicates
+            )
+
+        # ---------------------- Querying each plugin ---------------------- #
+        for plugin in plugins:
+            try:
+                logger.info(f"Loading plugin: {plugin.name} - v{plugin.dist.version}")  # noqa: G004
+
+                # Dynamically load the plugin class
+                plugin_class = plugin.load()
+
+                # Instantiate the plugin
+                self._plugins[plugin.name] = plugin_class()
+
+                # Store package distribution names for later use
+                self._dist_names[plugin.name] = plugin.dist.name
+            except Exception:
+                logging.exception("An unknown error happened - Ignoring plugin")
+
+    def get_provider_configs(self) -> dict[str, ProviderConfig]:
+        """Get a mapping of plugin names to provider configs"""
+
+        provider_cfgs = {}
+        for name, plugin_instance in self._plugins.items():
+            provider_cfg = plugin_instance.run()
+
+            if not isinstance(provider_cfg, ProviderConfig):
+                logging.error(
+                    f"Provider: {name} returned an unexpected type: "  # noqa: G004
+                    f"{type(provider_cfg)} - Expected: `ProviderConfig`. Ignoring..."
+                )
+                continue
+
+            provider_cfgs[name] = provider_cfg
+
+        return provider_cfgs
+
+    def get_dist_name_mapping(self) -> dict[str, str]:
+        """Get a mapping from plugin names to distribution names"""
+
+        return self._dist_names