Source code for gather.api

"""Gather -- Collect all your plugins

Gather allows a way to register plugins.
It features the ability to register the plugins from any module,
in any package, in any distribution.
A given module can register plugins of multiple types.

In order to have anything registered from a package,
it needs to declare that it supports :code:`gather`
in its package metadata.

For example,

.. code::

    ignored = "<ROOT_PACKAGE>"

The :code:`ROOT_PACKAGE` should point to the Python name of the package:
i.e., what users are expected to :code:`import` at the top-level.

Note that while having special facilities to run functions as subcommands,
Gather can be used to collect anything.
import collections
import contextlib
import importlib.metadata
import sys
import warnings

import attr
import venusian

def _ignore_deprecation():
    warnings.filterwarnings(action="ignore", category=DeprecationWarning)

def _get_modules():
    eps = importlib.metadata.entry_points()
    with _ignore_deprecation():
        gather_points = eps["gather"]
    for entry_point in gather_points:
        module = importlib.import_module(entry_point.value)
        yield module

[docs]@attr.s(frozen=True) class Collector(object): """ A plugin collector. A collector allows to *register* functions or classes by modules, and *collect*-ing them when they need to be used. """ name = attr.ib(default=None) depth = attr.ib(default=1)
[docs] def register(self, name=None, transform=lambda x: x): """ Register a class or function Args: name (str): optional. Name to register the class or function as. (default is name of object) transform (callable): optional. A one-argument function. Will be called, and the return value used in collection. Default is identity function This is meant to be used as a decoator: .. code:: @COLLECTOR.register() def specific_subcommand(args): pass @COLLECTOR.register(name='another_specific_name') def main(args): pass """ def callback(scanner, inner_name, objct): ( """ Venusian_ callback, called from scan .. _Venusian:""" """venusian/en/latest/api.html#venusian.attach """ ) tag = getattr(scanner, "tag", None) if tag is not self: return if name is None: effective_name = inner_name else: effective_name = name objct = transform(objct) scanner.registry[effective_name].add(objct) def attach(func): """Attach callback to be called when object is scanned""" venusian.attach(func, callback, depth=self.depth) return func return attach
[docs] def collect(self): """ Collect all registered functions or classes. Returns a dictionary mapping names to registered elements. """ def ignore_import_error(_unused): """ Ignore ImportError during collection. Some modules raise import errors for various reasons, and should be just treated as missing. """ if not issubclass(sys.exc_info()[0], ImportError): raise # pragma: no cover registry = collections.defaultdict(set) scanner = venusian.Scanner(registry=registry, tag=self) for module in _get_modules(): # Venusian is using a newly-deprecated method to scan modules with _ignore_deprecation(): scanner.scan(module, onerror=ignore_import_error) return registry
[docs]def unique(mapping): """ Transform map to sets to map to single items. Raises a :code:`ValueError` if any of the values is not an iterable with exactly one item. Args: mapping: A mapping of keys to Iterables of 1 Returns: A mapping of keys to the single value """ ret = {} for key, value_set in mapping.items(): [value] = value_set ret[key] = value return ret
[docs]@attr.s(frozen=True) class Wrapper(object): """Add extra data to an object""" original = attr.ib() extra = attr.ib()
[docs] @classmethod def glue(cls, extra): """ Glue extra data to an object Args: extra: what to add Returns: callable: function of one argument that returns a :code:`Wrapped` This method is useful mainly as the :code:`transform` parameter of a :code:`register` call. """ def ret(original): """Return a :code:`Wrapper` with the original and extra""" return cls(original=original, extra=extra) return ret
__all__ = ["Collector", "unique", "Wrapper"]