import functools
import importlib
import inspect
import os
import re
import sys
import warnings
from contextlib import contextmanager
from logging import getLogger
from pathlib import Path
from typing import (
Any,
Callable,
Dict,
Generator,
List,
Optional,
Set,
Tuple,
Type,
Union,
)
from . import _tracing
from .callers import HookResult
from .dist import (
_top_level_module_to_dist,
get_metadata,
importlib_metadata,
standard_metadata,
)
from .exceptions import (
Empty,
PluginError,
PluginImportError,
PluginRegistrationError,
PluginValidationError,
_empty,
)
from .hooks import HookCaller, HookExecFunc
from .implementation import HookImplementation, HookSpecification
from .markers import HookImplementationMarker, HookSpecificationMarker
logger = getLogger(__name__)
[docs]class PluginManager:
"""Core class which manages registration of plugin objects and hook calls.
You can register new hooks by calling :meth:`~PluginManager.add_hookspecs`.
You can register plugin objects (which contain hooks) by calling
:meth:`~PluginManager.register`. The ``PluginManager`` is initialized with
a ``project_name`` that is used when discovering *hook specifications* and
*hook implementations*.
For debugging purposes you may call :meth:`.PluginManager.enable_tracing`
which will subsequently send debug information to the trace helper.
Parameters
----------
project_name : str
The name of the host project. All :class:`HookImplementationMarker`
and :class:`HookSpecificationMarker` instances must be created using
the same ``project_name`` to be detected by this plugin manager.
discover_entry_point : str, optional
The default entry_point group to search when discovering plugins with
:meth:`PluginManager.discover`, by default None
discover_prefix : str, optional
The default module prefix to use when discovering plugins with
:meth:`PluginManager.discover`, by default None
discover_path : str or list of str, optional
A path or paths to include when discovering plugins with
:meth:`PluginManager.discover`, by default None
Examples
--------
.. code-block:: python
from napari_plugin_engine import PluginManager
import my_hookspecs
plugin_manager = PluginManager(
'my_project',
discover_entry_point='app.plugin',
discover_prefix='app_',
)
plugin_manager.add_hookspecs(my_hookspecs)
plugin_manager.discover()
# hooks now live in plugin_manager.hook
# plugin dict is at plugin_manager.plugins
"""
def __init__(
self,
project_name: str,
*,
discover_entry_point: Optional[str] = None,
discover_prefix: Optional[str] = None,
discover_path: Optional[List[str]] = None,
):
self.project_name = project_name
self.discover_entry_point = discover_entry_point
self.discover_prefix = discover_prefix
self.discover_path = discover_path or []
#: dict : mapping of ``plugin_name`` → ``plugin`` (object)
#:
#: Plugins get added to this dict in :meth:`~PluginManager.register`
self.plugins: Dict[str, Any] = {}
#: dict : mapping of ``plugin`` (object) → list of :class:`HookCaller`
#:
#: :class:`HookCaller` s get added in :meth:`~PluginManager.register`
self._plugin2hookcallers: Dict[Any, List[HookCaller]] = {}
self._blocked: Set[str] = set()
self.trace = _tracing.TagTracer().get("pluginmanage")
self.hook = _HookRelay(self)
self._inner_hookexec: HookExecFunc = lambda c, m, k: c.multicall(
m, k, firstresult=c.is_firstresult
)
@property
def hooks(self) -> '_HookRelay':
"""An alias for PluginManager.hook"""
return self.hook
def _hookexec(
self,
caller: HookCaller,
methods: List[HookImplementation],
kwargs: dict,
) -> HookResult:
"""Returns a function that will call a set of hookipmls with a caller.
This function will be passed to ``HookCaller`` instances that are
created during hookspec and plugin registration.
If :meth:`~.PluginManager.enable_tracing` is used, it will set it's own
wrapper function at self._inner_hookexec to enable tracing of hook
calls.
Parameters
----------
caller : HookCaller
The HookCaller instance that will call the HookImplementations.
methods : List[HookImplementation]
A list of :class:`~napari_plugin_engine.HookImplementation` objects
whose functions will be called during the hook call loop.
kwargs : dict
Keyword arguments to pass when calling the ``HookImplementation``.
Returns
-------
:class:`~napari_plugin_engine.HookResult`
The result object produced by the multicall loop.
"""
return self._inner_hookexec(caller, methods, kwargs)
[docs] def iter_available(
self,
path: Optional[str] = None,
entry_point: Optional[str] = None,
prefix: Optional[str] = None,
) -> Generator[Tuple[str, str, Optional[str]], None, None]:
"""Iterate over available plugins.
Parameters
----------
path : str, optional
If a string is provided, it is added to ``sys.path`` (and
``self.discover_path``) before importing, and removed at the end.
entry_point : str, optional
An entry_point group to search for, by default
``self.discover_entry_point`` is used
prefix : str, optional
If ``provided``, modules in the environment starting with
``prefix`` will be imported and searched for hook implementations
by default ``self.discover_prefix`` is used
See docstring of :func:`iter_available_plugins` for details.
"""
_path = self.discover_path
if path:
_path.append(path)
yield from iter_available_plugins(
entry_point or self.discover_entry_point,
prefix or self.discover_prefix,
_path,
include_uninstalled=bool(self.discover_prefix),
)
[docs] def discover(
self,
path: Optional[str] = None,
entry_point: Optional[str] = None,
prefix: Optional[str] = None,
ignore_errors: bool = True,
) -> Tuple[int, List[PluginError]]:
"""Discover and load plugins.
Parameters
----------
path : str, optional
If a string is provided, it is added to ``sys.path`` (and
``self.discover_path``) before importing, and removed at the end.
entry_point : str, optional
An entry_point group to search for, by default
``self.discover_entry_point`` is used
prefix : str, optional
If ``provided``, modules in the environment starting with
``prefix`` will be imported and searched for hook implementations
by default ``self.discover_prefix`` is used
ignore_errors : bool, optional
If ``True``, errors will be gathered and returned at the end.
Otherwise, they will be raised immediately. by default True
Returns
-------
(count, errs) : Tuple[int, List[PluginError]]
The number of succefully loaded modules, and a list of errors that
occurred (if ``ignore_errors`` was ``True``)
"""
self.hook._needs_discovery = False
# allow debugging escape hatch
if os.environ.get("DISABLE_ALL_PLUGINS"):
warnings.warn(
'Plugin discovery disabled due to '
'environmental variable "DISABLE_ALL_PLUGINS"'
)
return 0, []
errs: List[PluginError] = []
count = 0
for name, mod_name, dist_name in self.iter_available(
path, entry_point, prefix
):
if self.is_registered(name) or self.is_blocked(name):
continue
try:
if self._load_and_register(mod_name, name):
count += 1
except PluginError as e:
errs.append(e)
self.set_blocked(name)
if ignore_errors:
continue
raise e
return count, errs
[docs] @contextmanager
def discovery_blocked(self) -> Generator:
"""A context manager that temporarily blocks discovery of new plugins."""
current = self.hook._needs_discovery
self.hook._needs_discovery = False
try:
yield
finally:
self.hook._needs_discovery = current
def _load_and_register(
self, mod_name: str, plugin_name: Optional[str] = None
) -> Optional[str]:
"""A helper function to import and register a module as ``plugin_name``.
Parameters
----------
mod : str
The name of a module (or class in a module) to load.
plugin_name : str, optional
Optional name for plugin, by default ``get_canonical_name(plugin)``
Returns
-------
str or None
canonical plugin name, or ``None`` if the name is blocked from
registering.
Raises
------
PluginImportError
If an exception is raised when importing the module.
PluginRegistrationError
If an entry_point is declared that is neither a module nor a class.
PluginRegistrationError
If an exception is raised during plugin registration.
"""
try:
module = load(mod_name)
if self.is_registered(module):
return None
except Exception as exc:
raise PluginImportError(
f'Error while importing module {mod_name}',
plugin_name=plugin_name,
cause=exc,
)
if not (inspect.isclass(module) or inspect.ismodule(module)):
raise PluginRegistrationError(
f'Plugin "{plugin_name}" declared entry_point "{mod_name}"'
' which is neither a module nor a class.',
plugin=module,
plugin_name=plugin_name,
)
try:
return self.register(module, plugin_name)
except PluginError:
raise
except Exception as exc:
raise PluginRegistrationError(
plugin=module, plugin_name=plugin_name, cause=exc
)
[docs] def register(
self, namespace: Any, name: Optional[str] = None
) -> Optional[str]:
"""Register a plugin and return its canonical name or ``None``.
Parameters
----------
plugin : Any
The namespace (class, module, dict, etc...) to register
name : str, optional
Optional name for plugin, by default ``get_canonical_name(plugin)``
Returns
-------
str or None
canonical plugin name, or ``None`` if the name is blocked from
registering.
Raises
------
TypeError
If ``namespace`` is a string.
ValueError
if the plugin ``name`` or ``namespace`` is already registered.
"""
if isinstance(namespace, str):
raise TypeError("Plugin objects cannot be strings.")
if isinstance(namespace, dict):
return self._register_dict(namespace, name)
plugin_name = name or get_canonical_name(namespace)
if self.is_blocked(plugin_name):
return None
if self.is_registered(plugin_name):
raise ValueError(f"Plugin name already registered: {plugin_name}")
if self.is_registered(namespace):
raise ValueError(f"Plugin module already registered: {namespace}")
hookcallers = []
for hookimpl in iter_implementations(namespace, self.project_name):
hookimpl.plugin_name = plugin_name
hook_caller = getattr(self.hook, hookimpl.specname, None)
# if we don't yet have a hookcaller by this name, create one.
if hook_caller is None:
hook_caller = HookCaller(hookimpl.specname, self._hookexec)
setattr(self.hook, hookimpl.specname, hook_caller)
# otherwise, if it has a specification, validate the new
# hookimpl against the specification.
elif hook_caller.has_spec():
self._verify_hook(hook_caller, hookimpl)
hook_caller._maybe_apply_history(hookimpl)
# Finally, add the hookimpl to the hook_caller and the hook
# caller to the list of callers for this plugin.
hook_caller._add_hookimpl(hookimpl)
hookcallers.append(hook_caller)
self._plugin2hookcallers[namespace] = hookcallers
self.plugins[plugin_name] = namespace
return plugin_name
def _register_dict(
self, dct: Dict[str, Callable], name: Optional[str] = None, **kwargs
) -> Optional[str]:
"""Register a dict as a mapping of method name -> method.
Parameters
----------
dct : Dict[str, Callable]
Mapping of method name to method.
name : Optional[str], optional
The plugin_name to assign to this object, by default None
Returns
-------
str or None
canonical plugin name, or ``None`` if the name is blocked from
registering.
"""
mark = HookImplementationMarker(self.project_name)
clean_dct = {
key: mark(specname=key, **kwargs)(val)
for key, val in dct.items()
if inspect.isfunction(val)
}
namespace = ensure_namespace(clean_dct)
return self.register(namespace, name)
[docs] def get_name(self, plugin):
""" Return name for registered plugin or ``None`` if not registered. """
for name, val in self.plugins.items():
if plugin == val:
return name
def _ensure_plugin(self, name_or_object: Any) -> Any:
"""Return plugin object given a name or object. Or raise an exception.
Parameters
----------
name_or_object : Any
Either a string (in which case it is interpreted as a plugin name),
or a non-string object (in which case it is assumed to be a plugin
module or class).
Returns
-------
Any
The plugin object, if found.
Raises
------
KeyError
If the plugin does not exist.
"""
if isinstance(name_or_object, str):
plugin_name = name_or_object
else:
plugin_name = self.get_name(name_or_object)
if plugin_name in self.plugins:
return self.plugins[plugin_name]
if isinstance(name_or_object, str):
msg = f"No plugin found with the name {name_or_object}"
else:
msg = f"No plugin found with the name {name_or_object}"
raise KeyError(msg)
[docs] def unregister(self, name_or_object: Any) -> Optional[Any]:
"""Unregister a plugin object or ``plugin_name``.
Parameters
----------
name_or_object : str or Any
A module/class object or a plugin name (string).
Returns
-------
module : Any or None
The module object, or None if the ``name_or_object`` was not found.
"""
try:
plugin = self._ensure_plugin(name_or_object)
except KeyError as e:
warnings.warn(str(e))
return None
del self.plugins[self.get_name(plugin)]
for hookcaller in self._plugin2hookcallers.pop(plugin, []):
hookcaller._remove_plugin(plugin)
return plugin
[docs] def prune(self):
"""Unregister modules that can no longer be imported.
Useful if pip uninstall has been run during the session.
"""
_top_level_module_to_dist.cache_clear()
for plugin_module in list(self.plugins.values()):
try:
importlib.reload(plugin_module)
except ModuleNotFoundError:
self.unregister(plugin_module)
def _add_hookspec_dict(self, dct: Dict[str, Callable], **kwargs):
mark = HookSpecificationMarker(self.project_name)
clean_dct = {
key: mark(**kwargs)(val)
for key, val in dct.items()
if inspect.isfunction(val)
}
namespace = ensure_namespace(clean_dct)
return self.add_hookspecs(namespace)
[docs] def add_hookspecs(self, namespace: Any):
"""Add new hook specifications defined in the given ``namespace``.
Functions are recognized if they have been decorated accordingly.
"""
names = []
for name in dir(namespace):
method = getattr(namespace, name)
if not inspect.isroutine(method):
continue
tag = HookSpecification.format_tag(self.project_name)
spec_opts = getattr(method, tag, None)
if spec_opts is not None:
hook_caller = getattr(self.hook, name, None)
if hook_caller is None:
hook_caller = HookCaller(
name, self._hookexec, namespace, spec_opts
)
setattr(self.hook, name, hook_caller)
else:
# plugins registered this hook without knowing the spec
hook_caller.set_specification(namespace, spec_opts)
for hookfunction in hook_caller.get_hookimpls():
self._verify_hook(hook_caller, hookfunction)
names.append(name)
if not names:
raise ValueError(
f"did not find any {self.project_name!r} hooks in {namespace!r}"
)
[docs] def is_registered(self, obj: Any) -> bool:
"""Return ``True`` if the plugin is already registered."""
if isinstance(obj, str):
return obj in self.plugins
return obj in self._plugin2hookcallers
[docs] def is_blocked(self, plugin_name: str) -> bool:
"""Return ``True`` if the given plugin name is blocked."""
return plugin_name in self._blocked
[docs] def set_blocked(self, plugin_name: str, blocked=True):
"""Block registrations of ``plugin_name``, unregister if registered.
Parameters
----------
plugin_name : str
A plugin name to block.
blocked : bool, optional
Whether to block the plugin. If ``False`` will "unblock"
``plugin_name``. by default True
"""
if blocked:
self._blocked.add(plugin_name)
if self.is_registered(plugin_name):
self.unregister(plugin_name)
else:
if plugin_name in self._blocked:
self._blocked.remove(plugin_name)
[docs] def get_errors(
self,
plugin: Union[Any, Empty] = _empty,
error_type: Union[Type['PluginError'], Empty] = _empty,
) -> List[PluginError]:
"""Return a list of PluginErrors associated with ``plugin``.
Parameters
----------
plugin : Any
If provided, will restrict errors to those that were raised by
``plugin``. If a string is provided, it will be interpreted as the
name of the plugin, otherwise it is assumed to be the actual plugin
object itself.
error_type : PluginError
If provided, will restrict errors to instances of ``error_type``.
"""
# not using _ensure_plugin because it may not have been successfully
# registered
plugin_name: Union[str, Empty] = _empty
if isinstance(plugin, str):
plugin_name = plugin
plugin = _empty
return PluginError.get(
plugin=plugin, plugin_name=plugin_name, error_type=error_type
)
def _verify_hook(
self, hook_caller: HookCaller, hookimpl: HookImplementation
):
"""Check validity of a ``hookimpl``
Parameters
----------
hook_caller : HookCaller
A :class:`HookCaller` instance.
hookimpl : HookImplementation
A :class:`HookImplementation` instance, implementing the hook in
``hook_caller``.
Raises
------
PluginValidationError
If hook_caller is historic and the hookimpl is a hookwrapper.
PluginValidationError
If there are any argument names in the ``hookimpl`` that are not
in the ``hook_caller.spec``.
Warns
-----
Warning
If the hookspec has ``warn_on_impl`` flag (usually a deprecation).
"""
# historic hooks cannot have hookwrappers
if hook_caller.is_historic() and hookimpl.hookwrapper:
raise PluginValidationError(
hookimpl,
f"Plugin {hookimpl.plugin_name!r}\nhook "
f"{hook_caller.name!r}\nhistoric incompatible to hookwrapper",
)
if not hook_caller.spec:
return
# If the hookspec has ``warn_on_impl`` flag show a warning.
if hook_caller.spec.warn_on_impl:
warnings.warn_explicit(
hook_caller.spec.warn_on_impl,
type(hook_caller.spec.warn_on_impl),
lineno=hookimpl.function.__code__.co_firstlineno,
filename=hookimpl.function.__code__.co_filename,
)
# If there are any argument names in the hookimpl that are not
# in the hook specification.
notinspec = set(hookimpl.argnames) - set(hook_caller.spec.argnames)
if notinspec:
raise PluginValidationError(
hookimpl,
f"Plugin {hookimpl.plugin_name!r} for hook {hook_caller.name!r}"
f"\nhookimpl definition: {_formatdef(hookimpl.function)}\n"
f"Argument(s) {notinspec} are declared in the hookimpl but "
"can not be found in the hookspec",
)
[docs] def check_pending(self):
"""Make sure all hooks have a specification, or are optional.
Raises
------
PluginValidationError
If a hook implementation that was *not* marked as ``optionalhook``
has been registered for a non-existent hook specification.
"""
for name in self.hook.__dict__:
if name.startswith("_"):
continue
hook = getattr(self.hook, name)
if not hook.has_spec():
for hookimpl in hook.get_hookimpls():
if not hookimpl.optionalhook:
raise PluginValidationError(
hookimpl,
f"unknown hook {name!r} in "
f"plugin {hookimpl.plugin!r}",
)
[docs] def get_hookcallers(self, plugin: Any) -> Optional[List[HookCaller]]:
""" get all hook callers for the specified plugin. """
return self._plugin2hookcallers.get(plugin)
[docs] def add_hookcall_monitoring(
self,
before: Callable[[str, List[HookImplementation], dict], None],
after: Callable[
[HookResult, str, List[HookImplementation], dict], None
],
) -> Callable[[], None]:
"""Add before/after tracing functions for all hooks.
return an undo function which, when called, will remove the added
tracers.
``before(hook_name, hook_impls, kwargs)`` will be called ahead of all
hook calls and receive a hookcaller instance, a list of HookImplementation
instances and the keyword arguments for the hook call.
``after(outcome, hook_name, hook_impls, kwargs)`` receives the same
arguments as ``before`` but also a
:py:class:`napari_plugin_engine.callers._Result` object which
represents the result of the overall hook call.
"""
oldcall = self._inner_hookexec
def traced_hookexec(
caller: HookCaller, impls: List[HookImplementation], kwargs: dict
):
before(caller.name, impls, kwargs)
outcome = HookResult.from_call(
lambda: oldcall(caller, impls, kwargs)
)
after(outcome, caller.name, impls, kwargs)
return outcome
self._inner_hookexec = traced_hookexec
def undo():
self._inner_hookexec = oldcall
return undo
[docs] def enable_tracing(self):
"""Enable tracing of hook calls and return an undo function. """
hooktrace = self.trace.root.get("hook")
def before(hook_name, methods, kwargs):
hooktrace.root.indent += 1
hooktrace(hook_name, kwargs)
def after(outcome, hook_name, methods, kwargs):
if outcome.excinfo is None:
hooktrace("finish", hook_name, "-->", outcome.result)
hooktrace.root.indent -= 1
return self.add_hookcall_monitoring(before, after)
def __str__(self) -> str:
nhooks = len(self.hooks)
nplug = len(self.plugins)
text = f'PluginManager for "{self.project_name}"\n'
text += f'({nhooks} hook specs and {nplug} plugins)\n'
text += '-' * 45 + '\n'
for name, plugin in sorted(self.plugins.items(), key=lambda x: x[0]):
text += self.plugin_info(plugin) + "\n"
# nhooks = len(self._plugin2hookcallers[plugin])
# text += f'{name:29} {nhooks:3} hooks\n'
return text
def plugin_info(self, plugin) -> str:
plugin = self._ensure_plugin(plugin)
plugin_name = self.get_name(plugin)
version = self.get_metadata(plugin, 'version')
hooks = self._plugin2hookcallers[plugin]
name = f'{plugin_name} v{version}'
text = f'{name:34} {len(hooks):3} hooks\n'
for hook_caller in hooks:
for impl in hook_caller.get_hookimpls():
if impl.plugin_name == plugin_name:
text += f" - {impl.specname}\n"
return text
def _formatdef(func):
return f"{func.__name__}{str(inspect.signature(func))}"
class _HookRelay:
"""Hook holder object for storing HookCaller instances.
This object triggers (lazy) discovery of plugins as follows: When a plugin
hook is accessed (e.g. plugin_manager.hook.napari_get_reader), if
``self._needs_discovery`` is True, then it will trigger autodiscovery on
the parent plugin_manager. Note that ``PluginManager.__init__`` sets
``self.hook._needs_discovery = True`` *after* hook_specifications and
builtins have been discovered, but before external plugins are loaded.
"""
def __init__(self, manager: PluginManager):
self._manager = manager
self._needs_discovery = True
def __getattribute__(self, name) -> HookCaller:
"""Trigger manager plugin discovery when accessing hook first time."""
if name not in ("_needs_discovery", "_manager"):
if self._needs_discovery:
self._manager.discover()
return object.__getattribute__(self, name)
def __str__(self) -> str:
text = ''
for hookname, hookcaller in sorted(self.items(), key=lambda x: x[0]):
text += (
f'{hookname:25} {len(hookcaller.get_hookimpls()):3}'
' implementations\n'
)
return text
def __len__(self) -> int:
return len([k for k in vars(self) if not k.startswith("_")])
def items(self) -> List[Tuple[str, HookCaller]]:
"""Iterate through hookcallers, removing private attributes."""
return [
(k, val) for k, val in vars(self).items() if not k.startswith("_")
]
def values(self) -> List[HookCaller]:
"""Iterate through hookcallers, removing private attributes."""
return [val for k, val in vars(self).items() if not k.startswith("_")]
def get_canonical_name(namespace: Any) -> str:
"""Return canonical name for a plugin object.
Note that a plugin may be registered under a different name which was
specified by the caller of :meth:`PluginManager.register(plugin, name)
<.PluginManager.register>`. To obtain the name of a registered plugin
use :meth:`get_name(plugin) <.PluginManager.get_name>` instead.
"""
return getattr(namespace, "__name__", None) or str(id(namespace))
def iter_implementations(
namespace, project_name: str
) -> Generator[HookImplementation, None, None]:
# register matching hook implementations of the plugin
for name in dir(namespace):
# check all attributes/methods of plugin and look for functions or
# methods that have a "{self.project_name}_impl" attribute.
method = getattr(namespace, name)
if not inspect.isroutine(method):
continue
tag = HookImplementation.format_tag(project_name)
hookimpl_opts = getattr(method, tag, None)
if not (isinstance(hookimpl_opts, dict) and hookimpl_opts):
# false positive
continue
# create the HookImplementation instance for this method
try:
yield HookImplementation(method, namespace, **hookimpl_opts)
except TypeError:
# final fallback if the hookimpl_opts dict has invalid keys
# it's probably not a real hook implementation anyway.
pass
def ensure_namespace(obj: Any, name: str = 'orphan') -> Type:
"""Convert a ``dict`` to an object that provides ``getattr``.
Parameters
----------
obj : Any
An object, may be a ``dict``, or a regular namespace object.
name : str, optional
A name to use for the new namespace, if created. by default 'orphan'
Returns
-------
type
A namespace object. If ``obj`` is a ``dict``, creates a new ``type``
named ``name``, prepopulated with the key:value pairs from ``obj``.
Otherwise, if ``obj`` is not a ``dict``, will return the original
``obj``.
Raises
------
ValueError
If ``obj`` is a ``dict`` that contains keys that are not valid
`identifiers
<https://docs.python.org/3.3/reference/lexical_analysis.html#identifiers>`_.
"""
if isinstance(obj, dict):
bad_keys = [str(k) for k in obj.keys() if not str(k).isidentifier()]
if bad_keys:
raise ValueError(
f"dict contained invalid identifiers: {', '.join(bad_keys)}"
)
return type(name, (), obj)
return obj
@contextmanager
def temp_path_additions(path: Optional[Union[str, List[str]]]) -> Generator:
"""A context manager that temporarily adds ``path`` to sys.path.
Parameters
----------
path : str or list of str
A path or list of paths to add to sys.path
Yields
-------
sys_path : list of str
The current sys.path for the context.
"""
if isinstance(path, (str, Path)):
path = [path]
path = [os.fspath(p) for p in path] if path else []
to_add = [p for p in path if p not in sys.path]
for p in to_add:
sys.path.insert(0, p)
try:
yield sys.path
finally:
for p in to_add:
sys.path.remove(p)
pattern = re.compile(
r'(?P<module>[\w.]+)\s*'
r'(:\s*(?P<attr>[\w.]+))?\s*'
r'(?P<extras>\[.*\])?\s*$'
)
def load(value: str):
"""Load and return a module or attribute of a module (such as a class).
If only a module is indicated by the value, return that module. Otherwise,
return the named object.
"""
match = pattern.match(value)
if not match:
raise ValueError(f"malformed entry point string: {value}")
module = importlib.import_module(match.group('module'))
attrs = filter(None, (match.group('attr') or '').split('.'))
return functools.reduce(getattr, attrs, module)
def iter_available_plugins(
group: Optional[str] = None,
prefix: Optional[str] = None,
path: Optional[Union[str, List[str]]] = None,
include_uninstalled: bool = None,
) -> Generator[Tuple[str, str, Optional[str]], None, None]:
"""Discover modules by both naming convention and entry_points.
1. `Using naming convention
<https://packaging.python.org/guides/creating-and-discovering-plugins/#using-naming-convention>`_:
modules installed in the environment that follow a naming convention
(e.g. "napari_plugin"), can be discovered using :mod:`pkgutil`. This also
enables easy discovery using the PyPI `simple API
<https://www.python.org/dev/peps/pep-0503/>`_
2) `Using package metadata
<https://packaging.python.org/guides/creating-and-discovering-plugins/#using-package-metadata>`_:
packages that declare an `entry_point
<https://setuptools.readthedocs.io/en/latest/setuptools.html#dynamic-discovery-of-services-and-plugins>`_
in their ``setup.py`` file that matches the ``entry_point`` argument
can be discovered using `importlib.metadata
<https://docs.python.org/3/library/importlib.metadata.html>`_.
For background on entry points, see the `Entry Point specification
<https://packaging.python.org/specifications/entry-points/>`_.
Parameters
----------
group : str
The entry_point group name to search for
prefix : str
Any modules found in sys.path whose names begin with ``prefix``
will be imported and searched for hook implementations.
path : str or list of str, optional
Path or paths to add to sys.path before importing, removed at the end.
include_uninstalled : bool, optional
Whether to search for "local" (uninstalled) modules. Requires that a
prefix is provided. By default, True when prefix is provided.
Raises
------
ValueError
If ``include_uninstalled`` is true and ``prefix`` is not provided.
"""
if include_uninstalled is None:
include_uninstalled = bool(prefix)
with temp_path_additions(path):
_seen = set()
for dist in importlib_metadata.distributions():
matched = False
if group and not os.getenv("DISABLE_ENTRYPOINT_PLUGINS"):
for ep in dist.entry_points:
if ep.group == group: # type: ignore
matched = True
_seen.add(ep.value.split(".", maxsplit=1)[0])
yield (
ep.name,
ep.value,
dist.metadata.get("name"),
)
if matched:
continue
if prefix and not os.getenv("DISABLE_PREFIX_PLUGINS"):
name = dist.metadata.get("name")
if not name or name == prefix or (not name.startswith(prefix)):
continue
top_modules = dist.read_text('top_level.txt') or ""
for mod in filter(None, top_modules.split('\n')):
if mod.startswith(prefix):
_seen.add(mod)
yield (name, mod, name)
if include_uninstalled and not os.getenv("DISABLE_PREFIX_PLUGINS"):
from pkgutil import iter_modules
if not prefix:
raise ValueError(
"A prefix must be provided with 'include_uninstalled'."
)
for finder, mod_name, ispkg in iter_modules():
if (
mod_name.startswith(prefix)
and mod_name != prefix
and mod_name not in _seen
):
yield (mod_name, mod_name, None)