move PluginManager to src/ape/managers/plugins.py and clean_plugin_na…

…me to ape/plugins/_utils to avoid circular import

src/ape/_cli.py

@@ -11,8 +11,7 @@
 from ape.cli import ape_cli_context
 from ape.exceptions import Abort, ApeException, handle_ape_exception
 from ape.logging import logger
-from ape.plugins import clean_plugin_name
-from ape.plugins._utils import PluginMetadataList
+from ape.plugins._utils import PluginMetadataList, clean_plugin_name
 from ape.utils.basemodel import ManagerAccessMixin
 
 _DIFFLIB_CUT_OFF = 0.6

src/ape/api/networks.py

@@ -966,10 +966,10 @@ def providers(self):  # -> Dict[str, Partial[ProviderAPI]]
             Dict[str, partial[:class:`~ape.api.providers.ProviderAPI`]]
         """
 
-        from ape.plugins import clean_plugin_name
+        from ape.plugins._utils import clean_plugin_name
 
         providers = {}
-        for plugin_name, plugin_tuple in self.plugin_manager.providers:
+        for _, plugin_tuple in self.plugin_manager.providers:
             ecosystem_name, network_name, provider_class = plugin_tuple
             provider_name = clean_plugin_name(provider_class.__module__.split(".")[0])
 

src/ape/managers/plugins.py

@@ -0,0 +1,154 @@
+import importlib
+import pkgutil
+import subprocess
+from typing import Generator, Iterator, List, Optional, Set, Tuple
+
+from ape.__modules__ import __modules__
+from ape.exceptions import ApeAttributeError
+from ape.logging import logger
+from ape.plugins._utils import PIP_COMMAND, clean_plugin_name
+from ape.utils.basemodel import _assert_not_ipython_check
+from ape.utils.misc import log_instead_of_fail
+
+
+class PluginManager:
+    _unimplemented_plugins: List[str] = []
+
+    def __init__(self) -> None:
+        self.__registered = False
+
+    @log_instead_of_fail(default="<PluginManager>")
+    def __repr__(self) -> str:
+        return f"<{PluginManager.__name__}>"
+
+    def __getattr__(self, attr_name: str) -> Iterator[Tuple[str, Tuple]]:
+        _assert_not_ipython_check(attr_name)
+
+        # NOTE: The first time this method is called, the actual
+        #  plugin registration occurs. Registration only happens once.
+        self._register_plugins()
+
+        if not hasattr(pluggy_manager.hook, attr_name):
+            raise ApeAttributeError(f"{PluginManager.__name__} has no attribute '{attr_name}'.")
+
+        # Do this to get access to the package name
+        hook_fn = getattr(pluggy_manager.hook, attr_name)
+        hookimpls = hook_fn.get_hookimpls()
+
+        def get_plugin_name_and_hookfn(h):
+            return h.plugin_name, getattr(h.plugin, attr_name)()
+
+        for plugin_name, results in map(get_plugin_name_and_hookfn, hookimpls):
+            # NOTE: Some plugins return a tuple and some return iterators
+            if not isinstance(results, Generator):
+                validated_plugin = self._validate_plugin(plugin_name, results)
+                if validated_plugin:
+                    yield validated_plugin
+            else:
+                # Only if it's an iterator, provider results as a series
+                for result in results:
+                    validated_plugin = self._validate_plugin(plugin_name, result)
+                    if validated_plugin:
+                        yield validated_plugin
+
+    @property
+    def registered_plugins(self) -> Set[str]:
+        self._register_plugins()
+        return {x[0] for x in pluggy_manager.list_name_plugin()}
+
+    @functools.cached_property
+    def _plugin_modules(self) -> Tuple[str, ...]:
+        core_plugin_module_names = {n for _, n, _ in pkgutil.iter_modules() if n.startswith("ape_")}
+        if PIP_COMMAND[0] != "uv":
+            # NOTE: Unable to use pkgutil.iter_modules() for installed plugins
+            # because it does not work with editable installs.
+            # See https://github.com/python/cpython/issues/99805.
+            result = subprocess.check_output(
+                ["pip", "list", "--format", "freeze", "--disable-pip-version-check"]
+            )
+            packages = result.decode("utf8").splitlines()
+            installed_plugin_module_names = {
+                p.split("==")[0].replace("-", "_") for p in packages if p.startswith("ape-")
+            }
+        else:
+            result = subprocess.check_output(["uv", "pip", "list"])
+            # format is in the output of:
+            # Package                   Version                         Editable project location
+            # ------------------------- ------------------------------- -------------------------
+            # aiosignal                 1.3.1
+            # annotated-types           0.6.0
+            # skip the header
+            packages = result.decode("utf8").splitlines()[2:]
+            installed_plugin_module_names = {
+                p.split(" ")[0].replace("-", "_") for p in packages if p.startswith("ape-")
+            }
+
+        # NOTE: Returns tuple because this shouldn't change.
+        return tuple(installed_plugin_module_names.union(core_plugin_module_names))
+
+    def _register_plugins(self):
+        if self.__registered:
+            return
+
+        for module_name in self._plugin_modules:
+            try:
+                module = importlib.import_module(module_name)
+                pluggy_manager.register(module)
+            except Exception as err:
+                if module_name in __modules__:
+                    # Always raise core plugin registration errors.
+                    raise
+
+                logger.warn_from_exception(err, f"Error loading plugin package '{module_name}'.")
+
+        self.__registered = True
+
+    def _validate_plugin(self, plugin_name: str, plugin_cls) -> Optional[Tuple[str, Tuple]]:
+        if valid_impl(plugin_cls):
+            return clean_plugin_name(plugin_name), plugin_cls
+        else:
+            self._warn_not_fully_implemented_error(plugin_cls, plugin_name)
+            return None
+
+    def _warn_not_fully_implemented_error(self, results, plugin_name):
+        if plugin_name in self._unimplemented_plugins:
+            # Already warned
+            return
+
+        unimplemented_methods = []
+
+        # Find the best API name to warn about.
+        if isinstance(results, (list, tuple)):
+            classes = [p for p in results if hasattr(p, "__name__")]
+            if classes:
+                # Likely only ever a single class in a registration, but just in case.
+                api_name = " - ".join([p.__name__ for p in classes if hasattr(p, "__name__")])
+                for api_cls in classes:
+                    if (
+                        abstract_methods := getattr(api_cls, "__abstractmethods__", None)
+                    ) and isinstance(abstract_methods, dict):
+                        unimplemented_methods.extend(api_cls.__abstractmethods__)
+
+            else:
+                # This would only happen if the registration consisted of all primitives.
+                api_name = " - ".join(results)
+
+        elif hasattr(results, "__name__"):
+            api_name = results.__name__
+            if (abstract_methods := getattr(results, "__abstractmethods__", None)) and isinstance(
+                abstract_methods, dict
+            ):
+                unimplemented_methods.extend(results.__abstractmethods__)
+
+        else:
+            api_name = results
+
+        message = f"'{api_name}' from '{plugin_name}' is not fully implemented."
+        if unimplemented_methods:
+            methods_str = ", ".join(unimplemented_methods)
+            message = f"{message} Remaining abstract methods: '{methods_str}'."
+
+        logger.warning(message)
+
+        # Record so we don't warn repeatedly
+        self._unimplemented_plugins.append(plugin_name)

src/ape/managers/query.py

@@ -15,7 +15,7 @@
 from ape.contracts.base import ContractLog, LogFilter
 from ape.exceptions import QueryEngineError
 from ape.logging import logger
-from ape.plugins import clean_plugin_name
+from ape.plugins._utils import clean_plugin_name
 from ape.utils import ManagerAccessMixin, cached_property, singledispatchmethod
 
 

src/ape/plugins/__init__.py

@@ -1,16 +1,8 @@
 import functools
-import importlib
-import pkgutil
-import subprocess
-from typing import Any, Callable, Generator, Iterator, List, Optional, Set, Tuple, Type
-
-from ape.__modules__ import __modules__
-from ape.exceptions import ApeAttributeError
-from ape.logging import logger
-from ape.utils.basemodel import _assert_not_ipython_check
-from ape.utils.misc import log_instead_of_fail
-
-from ._utils import PIP_COMMAND
+from typing import Any, Callable, Type
+
+from ape.managers.plugins import PluginManager
+
 from .account import AccountPlugin
 from .compiler import CompilerPlugin
 from .config import Config
@@ -47,10 +39,6 @@ class AllPluginHooks(
 pluggy_manager.add_hookspecs(AllPluginHooks)
 
 
-def clean_plugin_name(name: str) -> str:
-    return name.replace("_", "-").replace("ape-", "")
-
-
 def get_hooks(plugin_type):
     return [name for name, method in plugin_type.__dict__.items() if hasattr(method, "ape_spec")]
 
@@ -121,151 +109,7 @@ def valid_impl(api_class: Any) -> bool:
     return len(api_class.__abstractmethods__) == 0
 
 
-class PluginManager:
-    _unimplemented_plugins: List[str] = []
-
-    def __init__(self) -> None:
-        self.__registered = False
-
-    @log_instead_of_fail(default="<PluginManager>")
-    def __repr__(self) -> str:
-        return f"<{PluginManager.__name__}>"
-
-    def __getattr__(self, attr_name: str) -> Iterator[Tuple[str, Tuple]]:
-        _assert_not_ipython_check(attr_name)
-
-        # NOTE: The first time this method is called, the actual
-        #  plugin registration occurs. Registration only happens once.
-        self._register_plugins()
-
-        if not hasattr(pluggy_manager.hook, attr_name):
-            raise ApeAttributeError(f"{PluginManager.__name__} has no attribute '{attr_name}'.")
-
-        # Do this to get access to the package name
-        hook_fn = getattr(pluggy_manager.hook, attr_name)
-        hookimpls = hook_fn.get_hookimpls()
-
-        def get_plugin_name_and_hookfn(h):
-            return h.plugin_name, getattr(h.plugin, attr_name)()
-
-        for plugin_name, results in map(get_plugin_name_and_hookfn, hookimpls):
-            # NOTE: Some plugins return a tuple and some return iterators
-            if not isinstance(results, Generator):
-                validated_plugin = self._validate_plugin(plugin_name, results)
-                if validated_plugin:
-                    yield validated_plugin
-            else:
-                # Only if it's an iterator, provider results as a series
-                for result in results:
-                    validated_plugin = self._validate_plugin(plugin_name, result)
-                    if validated_plugin:
-                        yield validated_plugin
-
-    @property
-    def registered_plugins(self) -> Set[str]:
-        self._register_plugins()
-        return {x[0] for x in pluggy_manager.list_name_plugin()}
-
-    @functools.cached_property
-    def _plugin_modules(self) -> Tuple[str, ...]:
-        core_plugin_module_names = {n for _, n, _ in pkgutil.iter_modules() if n.startswith("ape_")}
-        if PIP_COMMAND[0] != "uv":
-            # NOTE: Unable to use pkgutil.iter_modules() for installed plugins
-            # because it does not work with editable installs.
-            # See https://github.com/python/cpython/issues/99805.
-            result = subprocess.check_output(
-                ["pip", "list", "--format", "freeze", "--disable-pip-version-check"]
-            )
-            packages = result.decode("utf8").splitlines()
-            installed_plugin_module_names = {
-                p.split("==")[0].replace("-", "_") for p in packages if p.startswith("ape-")
-            }
-        else:
-            result = subprocess.check_output(["uv", "pip", "list"])
-            # format is in the output of:
-            # Package                   Version                         Editable project location
-            # ------------------------- ------------------------------- -------------------------
-            # aiosignal                 1.3.1
-            # annotated-types           0.6.0
-            # skip the header
-            packages = result.decode("utf8").splitlines()[2:]
-            installed_plugin_module_names = {
-                p.split(" ")[0].replace("-", "_") for p in packages if p.startswith("ape-")
-            }
-
-        # NOTE: Returns tuple because this shouldn't change.
-        return tuple(installed_plugin_module_names.union(core_plugin_module_names))
-
-    def _register_plugins(self):
-        if self.__registered:
-            return
-
-        for module_name in self._plugin_modules:
-            try:
-                module = importlib.import_module(module_name)
-                pluggy_manager.register(module)
-            except Exception as err:
-                if module_name in __modules__:
-                    # Always raise core plugin registration errors.
-                    raise
-
-                logger.warn_from_exception(err, f"Error loading plugin package '{module_name}'.")
-
-        self.__registered = True
-
-    def _validate_plugin(self, plugin_name: str, plugin_cls) -> Optional[Tuple[str, Tuple]]:
-        if valid_impl(plugin_cls):
-            return clean_plugin_name(plugin_name), plugin_cls
-        else:
-            self._warn_not_fully_implemented_error(plugin_cls, plugin_name)
-            return None
-
-    def _warn_not_fully_implemented_error(self, results, plugin_name):
-        if plugin_name in self._unimplemented_plugins:
-            # Already warned
-            return
-
-        unimplemented_methods = []
-
-        # Find the best API name to warn about.
-        if isinstance(results, (list, tuple)):
-            classes = [p for p in results if hasattr(p, "__name__")]
-            if classes:
-                # Likely only ever a single class in a registration, but just in case.
-                api_name = " - ".join([p.__name__ for p in classes if hasattr(p, "__name__")])
-                for api_cls in classes:
-                    if (
-                        abstract_methods := getattr(api_cls, "__abstractmethods__", None)
-                    ) and isinstance(abstract_methods, dict):
-                        unimplemented_methods.extend(api_cls.__abstractmethods__)
-
-            else:
-                # This would only happen if the registration consisted of all primitives.
-                api_name = " - ".join(results)
-
-        elif hasattr(results, "__name__"):
-            api_name = results.__name__
-            if (abstract_methods := getattr(results, "__abstractmethods__", None)) and isinstance(
-                abstract_methods, dict
-            ):
-                unimplemented_methods.extend(results.__abstractmethods__)
-
-        else:
-            api_name = results
-
-        message = f"'{api_name}' from '{plugin_name}' is not fully implemented."
-        if unimplemented_methods:
-            methods_str = ", ".join(unimplemented_methods)
-            message = f"{message} Remaining abstract methods: '{methods_str}'."
-
-        logger.warning(message)
-
-        # Record so we don't warn repeatedly
-        self._unimplemented_plugins.append(plugin_name)
-
-
 __all__ = [
     "PluginManager",
-    "clean_plugin_name",
     "register",
 ]

src/ape/plugins/_utils.py

@@ -11,7 +11,6 @@
 
 from ape.__modules__ import __modules__
 from ape.logging import logger
-from ape.plugins import clean_plugin_name
 from ape.utils import BaseInterfaceModel, get_package_version, github_client, log_instead_of_fail
 from ape.utils.basemodel import BaseModel
 from ape.utils.misc import _get_distributions
@@ -24,6 +23,10 @@
 PIP_COMMAND = ["uv", "pip"] if which("uv") else [sys.executable, "-m", "pip"]
 
 
+def clean_plugin_name(name: str) -> str:
+    return name.replace("_", "-").replace("ape-", "")
+
+
 class ApeVersion:
     def __str__(self) -> str:
         return str(self.version)
@@ -211,9 +214,6 @@ def validate_name(cls, values):
                 # Just some small validation so you can't put a repo
                 # that isn't this plugin here. NOTE: Forks should still work.
                 raise ValueError("Plugin mismatch with remote git version.")
-
-            version = version
-
         elif not version:
             # Only check name for version constraint if not in version.
             # NOTE: This happens when using the CLI to provide version constraints.