Setting up deprecations of Pluggy Hooks

deprecate.py

import sys
import warnings
import inspect
from pluggy import HookimplMarker, HookspecMarker, PluginManager, _HookCaller, normalize_hookimpl_opts, HookImpl

class MetadataHookspecMarker(HookspecMarker):
    """
    Allows storing arbitrary metadata on the hookspec options
    instead of what Pluggy sets by default.
    """
    def __call__(self, function=None, firstresult=False, historic=False, **kwargs):
        def set_other_spec_opts(func):
            data = getattr(func, self.project_name + '_spec')
            data.update(kwargs)
            return func
        
        
        result = super(MetadataHookspecMarker, self).__call__(function, firstresult, historic)

        if function is not None:
            return set_other_spec_opts(result)
        else:
            return lambda function: set_other_spec_opts(result(function))

class DeprecatedHookCaller(_HookCaller):
    def __call__(self, *args, **kwargs):
        if args:
            raise TypeError("hook calling supports only keyword arguments")
            
        assert not self.is_historic()
        
        if self.argnames:
            notincall = set(self.argnames) - set(['__multicall__']) - set(kwargs.keys())
            if notincall:
                warnings.warn(
                    "Argument(s) {} which are declared in the hookspec "
                    "can not be found in this hook call".format(tuple(notincall)),
                    stacklevel=2,
                )
                
        self.precall()
        return self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
    
    def precall(self):
        deprecation = self.spec_opts.get('deprecated')
        if deprecation:
            warnings.warn(
                "{} will be deprecated in version {}".format(self.name, deprecation)
            )
        else:
            raise Exception("notice me senpai")

class InstrumentedManager(PluginManager):
    def __init__(self, project_name, implprefix=None, hook_caller_class=_HookCaller):
        super().__init__(project_name, implprefix)
        self.hook_caller_class = hook_caller_class


    def add_hookspecs(self, module_or_class):
        """ add new hook specifications defined in the given module_or_class.
        Functions are recognized if they have been decorated accordingly. """
        names = []
        for name in dir(module_or_class):
            spec_opts = self.parse_hookspec_opts(module_or_class, name)
            if spec_opts is not None:
                hc = getattr(self.hook, name, None)
                if hc is None:
                    hc = self.hook_caller_class(name, self._hookexec, module_or_class, spec_opts)
                    setattr(self.hook, name, hc)
                else:
                    # plugins registered this hook without knowing the spec
                    hc.set_specification(module_or_class, spec_opts)
                    for hookfunction in (hc._wrappers + hc._nonwrappers):
                        self._verify_hook(hc, hookfunction)
                names.append(name)

        if not names:
            raise ValueError("did not find any %r hooks in %r" % (self.project_name, module_or_class))


    def subset_hook_caller(self, name, remove_plugins):
        """ Return a new Caller instance for the named method
        which manages calls to all registered plugins except the
        ones from remove_plugins. """
        orig = getattr(self.hook, name)
        plugins_to_remove = [plug for plug in remove_plugins if hasattr(plug, name)]
        if plugins_to_remove:
            hc = self.hook_caller_class(orig.name, orig._hookexec, orig._specmodule_or_class,
                             orig.spec_opts)
            for hookimpl in (orig._wrappers + orig._nonwrappers):
                plugin = hookimpl.plugin
                if plugin not in plugins_to_remove:
                    hc._add_hookimpl(hookimpl)
                    # we also keep track of this hook caller so it
                    # gets properly removed on plugin unregistration
                    self._plugin2hookcallers.setdefault(plugin, []).append(hc)
            return hc
        return orig
    
    def register(self, plugin, name=None):
        """ Register a plugin and return its canonical name or None if the name
        is blocked from registering.  Raise a ValueError if the plugin is already
        registered. """
        plugin_name = name or self.get_canonical_name(plugin)

        if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers:
            if self._name2plugin.get(plugin_name, -1) is None:
                return  # blocked plugin, return None to indicate no registration
            raise ValueError("Plugin already registered: %s=%s" %
                             (plugin_name, plugin, self._name2plugin))

        # XXX if an error happens we should make sure no state has been
        # changed at point of return
        self._name2plugin[plugin_name] = plugin

        # register matching hook implementations of the plugin
        self._plugin2hookcallers[plugin] = hookcallers = []
        for name in dir(plugin):
            hookimpl_opts = self.parse_hookimpl_opts(plugin, name)
            if hookimpl_opts is not None:
                normalize_hookimpl_opts(hookimpl_opts)
                method = getattr(plugin, name)
                hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)
                hook = getattr(self.hook, name, None)
                if hook is None:
                    hook = self.hook_caller_class(name, self._hookexec)
                    setattr(self.hook, name, hook)
                elif hook.has_spec():
                    self._verify_hook(hook, hookimpl)
                    hook._maybe_apply_history(hookimpl)
                hook._add_hookimpl(hookimpl)
                hookcallers.append(hook)
        return plugin_name

pm = InstrumentedManager('test', hook_caller_class=DeprecatedHookCaller)
spec = MetadataHookspecMarker('test')


@spec(deprecated='1.0')
def some_hook():
    pass

pm.add_hookspecs(sys.modules[__name__])


@impl
def some_hook():
    pass

pm.register(sys.modules[__name__])


pm.hook.some_hook()