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 importlib.metadata
import sys

import attr
import venusian

def _get_modules():
    for entry_point in importlib.metadata.entry_points(group="gather"):
        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(): 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"]