Spaces:
Running
Running
| """ | |
| Internal hook annotation, representation and calling machinery. | |
| """ | |
| import inspect | |
| import sys | |
| import warnings | |
| class HookspecMarker: | |
| """Decorator helper class for marking functions as hook specifications. | |
| You can instantiate it with a project_name to get a decorator. | |
| Calling :py:meth:`.PluginManager.add_hookspecs` later will discover all marked functions | |
| if the :py:class:`.PluginManager` uses the same project_name. | |
| """ | |
| def __init__(self, project_name): | |
| self.project_name = project_name | |
| def __call__( | |
| self, function=None, firstresult=False, historic=False, warn_on_impl=None | |
| ): | |
| """if passed a function, directly sets attributes on the function | |
| which will make it discoverable to :py:meth:`.PluginManager.add_hookspecs`. | |
| If passed no function, returns a decorator which can be applied to a function | |
| later using the attributes supplied. | |
| If ``firstresult`` is ``True`` the 1:N hook call (N being the number of registered | |
| hook implementation functions) will stop at I<=N when the I'th function | |
| returns a non-``None`` result. | |
| If ``historic`` is ``True`` calls to a hook will be memorized and replayed | |
| on later registered plugins. | |
| """ | |
| def setattr_hookspec_opts(func): | |
| if historic and firstresult: | |
| raise ValueError("cannot have a historic firstresult hook") | |
| setattr( | |
| func, | |
| self.project_name + "_spec", | |
| dict( | |
| firstresult=firstresult, | |
| historic=historic, | |
| warn_on_impl=warn_on_impl, | |
| ), | |
| ) | |
| return func | |
| if function is not None: | |
| return setattr_hookspec_opts(function) | |
| else: | |
| return setattr_hookspec_opts | |
| class HookimplMarker: | |
| """Decorator helper class for marking functions as hook implementations. | |
| You can instantiate with a ``project_name`` to get a decorator. | |
| Calling :py:meth:`.PluginManager.register` later will discover all marked functions | |
| if the :py:class:`.PluginManager` uses the same project_name. | |
| """ | |
| def __init__(self, project_name): | |
| self.project_name = project_name | |
| def __call__( | |
| self, | |
| function=None, | |
| hookwrapper=False, | |
| optionalhook=False, | |
| tryfirst=False, | |
| trylast=False, | |
| specname=None, | |
| ): | |
| """if passed a function, directly sets attributes on the function | |
| which will make it discoverable to :py:meth:`.PluginManager.register`. | |
| If passed no function, returns a decorator which can be applied to a | |
| function later using the attributes supplied. | |
| If ``optionalhook`` is ``True`` a missing matching hook specification will not result | |
| in an error (by default it is an error if no matching spec is found). | |
| If ``tryfirst`` is ``True`` this hook implementation will run as early as possible | |
| in the chain of N hook implementations for a specification. | |
| If ``trylast`` is ``True`` this hook implementation will run as late as possible | |
| in the chain of N hook implementations. | |
| If ``hookwrapper`` is ``True`` the hook implementations needs to execute exactly | |
| one ``yield``. The code before the ``yield`` is run early before any non-hookwrapper | |
| function is run. The code after the ``yield`` is run after all non-hookwrapper | |
| function have run. The ``yield`` receives a :py:class:`.callers._Result` object | |
| representing the exception or result outcome of the inner calls (including other | |
| hookwrapper calls). | |
| If ``specname`` is provided, it will be used instead of the function name when | |
| matching this hook implementation to a hook specification during registration. | |
| """ | |
| def setattr_hookimpl_opts(func): | |
| setattr( | |
| func, | |
| self.project_name + "_impl", | |
| dict( | |
| hookwrapper=hookwrapper, | |
| optionalhook=optionalhook, | |
| tryfirst=tryfirst, | |
| trylast=trylast, | |
| specname=specname, | |
| ), | |
| ) | |
| return func | |
| if function is None: | |
| return setattr_hookimpl_opts | |
| else: | |
| return setattr_hookimpl_opts(function) | |
| def normalize_hookimpl_opts(opts): | |
| opts.setdefault("tryfirst", False) | |
| opts.setdefault("trylast", False) | |
| opts.setdefault("hookwrapper", False) | |
| opts.setdefault("optionalhook", False) | |
| opts.setdefault("specname", None) | |
| _PYPY = hasattr(sys, "pypy_version_info") | |
| def varnames(func): | |
| """Return tuple of positional and keywrord argument names for a function, | |
| method, class or callable. | |
| In case of a class, its ``__init__`` method is considered. | |
| For methods the ``self`` parameter is not included. | |
| """ | |
| if inspect.isclass(func): | |
| try: | |
| func = func.__init__ | |
| except AttributeError: | |
| return (), () | |
| elif not inspect.isroutine(func): # callable object? | |
| try: | |
| func = getattr(func, "__call__", func) | |
| except Exception: | |
| return (), () | |
| try: # func MUST be a function or method here or we won't parse any args | |
| spec = inspect.getfullargspec(func) | |
| except TypeError: | |
| return (), () | |
| args, defaults = tuple(spec.args), spec.defaults | |
| if defaults: | |
| index = -len(defaults) | |
| args, kwargs = args[:index], tuple(args[index:]) | |
| else: | |
| kwargs = () | |
| # strip any implicit instance arg | |
| # pypy3 uses "obj" instead of "self" for default dunder methods | |
| implicit_names = ("self",) if not _PYPY else ("self", "obj") | |
| if args: | |
| if inspect.ismethod(func) or ( | |
| "." in getattr(func, "__qualname__", ()) and args[0] in implicit_names | |
| ): | |
| args = args[1:] | |
| return args, kwargs | |
| class _HookRelay: | |
| """hook holder object for performing 1:N hook calls where N is the number | |
| of registered plugins. | |
| """ | |
| class _HookCaller: | |
| def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None): | |
| self.name = name | |
| self._wrappers = [] | |
| self._nonwrappers = [] | |
| self._hookexec = hook_execute | |
| self._call_history = None | |
| self.spec = None | |
| if specmodule_or_class is not None: | |
| assert spec_opts is not None | |
| self.set_specification(specmodule_or_class, spec_opts) | |
| def has_spec(self): | |
| return self.spec is not None | |
| def set_specification(self, specmodule_or_class, spec_opts): | |
| assert not self.has_spec() | |
| self.spec = HookSpec(specmodule_or_class, self.name, spec_opts) | |
| if spec_opts.get("historic"): | |
| self._call_history = [] | |
| def is_historic(self): | |
| return self._call_history is not None | |
| def _remove_plugin(self, plugin): | |
| def remove(wrappers): | |
| for i, method in enumerate(wrappers): | |
| if method.plugin == plugin: | |
| del wrappers[i] | |
| return True | |
| if remove(self._wrappers) is None: | |
| if remove(self._nonwrappers) is None: | |
| raise ValueError(f"plugin {plugin!r} not found") | |
| def get_hookimpls(self): | |
| # Order is important for _hookexec | |
| return self._nonwrappers + self._wrappers | |
| def _add_hookimpl(self, hookimpl): | |
| """Add an implementation to the callback chain.""" | |
| if hookimpl.hookwrapper: | |
| methods = self._wrappers | |
| else: | |
| methods = self._nonwrappers | |
| if hookimpl.trylast: | |
| methods.insert(0, hookimpl) | |
| elif hookimpl.tryfirst: | |
| methods.append(hookimpl) | |
| else: | |
| # find last non-tryfirst method | |
| i = len(methods) - 1 | |
| while i >= 0 and methods[i].tryfirst: | |
| i -= 1 | |
| methods.insert(i + 1, hookimpl) | |
| def __repr__(self): | |
| return f"<_HookCaller {self.name!r}>" | |
| def __call__(self, *args, **kwargs): | |
| if args: | |
| raise TypeError("hook calling supports only keyword arguments") | |
| assert not self.is_historic() | |
| # This is written to avoid expensive operations when not needed. | |
| if self.spec: | |
| for argname in self.spec.argnames: | |
| if argname not in kwargs: | |
| notincall = tuple(set(self.spec.argnames) - kwargs.keys()) | |
| warnings.warn( | |
| "Argument(s) {} which are declared in the hookspec " | |
| "can not be found in this hook call".format(notincall), | |
| stacklevel=2, | |
| ) | |
| break | |
| firstresult = self.spec.opts.get("firstresult") | |
| else: | |
| firstresult = False | |
| return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult) | |
| def call_historic(self, result_callback=None, kwargs=None): | |
| """Call the hook with given ``kwargs`` for all registered plugins and | |
| for all plugins which will be registered afterwards. | |
| If ``result_callback`` is not ``None`` it will be called for for each | |
| non-``None`` result obtained from a hook implementation. | |
| """ | |
| self._call_history.append((kwargs or {}, result_callback)) | |
| # Historizing hooks don't return results. | |
| # Remember firstresult isn't compatible with historic. | |
| res = self._hookexec(self.name, self.get_hookimpls(), kwargs, False) | |
| if result_callback is None: | |
| return | |
| for x in res or []: | |
| result_callback(x) | |
| def call_extra(self, methods, kwargs): | |
| """Call the hook with some additional temporarily participating | |
| methods using the specified ``kwargs`` as call parameters.""" | |
| old = list(self._nonwrappers), list(self._wrappers) | |
| for method in methods: | |
| opts = dict(hookwrapper=False, trylast=False, tryfirst=False) | |
| hookimpl = HookImpl(None, "<temp>", method, opts) | |
| self._add_hookimpl(hookimpl) | |
| try: | |
| return self(**kwargs) | |
| finally: | |
| self._nonwrappers, self._wrappers = old | |
| def _maybe_apply_history(self, method): | |
| """Apply call history to a new hookimpl if it is marked as historic.""" | |
| if self.is_historic(): | |
| for kwargs, result_callback in self._call_history: | |
| res = self._hookexec(self.name, [method], kwargs, False) | |
| if res and result_callback is not None: | |
| result_callback(res[0]) | |
| class HookImpl: | |
| def __init__(self, plugin, plugin_name, function, hook_impl_opts): | |
| self.function = function | |
| self.argnames, self.kwargnames = varnames(self.function) | |
| self.plugin = plugin | |
| self.opts = hook_impl_opts | |
| self.plugin_name = plugin_name | |
| self.__dict__.update(hook_impl_opts) | |
| def __repr__(self): | |
| return f"<HookImpl plugin_name={self.plugin_name!r}, plugin={self.plugin!r}>" | |
| class HookSpec: | |
| def __init__(self, namespace, name, opts): | |
| self.namespace = namespace | |
| self.function = function = getattr(namespace, name) | |
| self.name = name | |
| self.argnames, self.kwargnames = varnames(function) | |
| self.opts = opts | |
| self.warn_on_impl = opts.get("warn_on_impl") | |