Pluggy-Hookimpl的参数说明

function=func

通过直接传入func来设置function attribute，如果为None，则返回一个包装器 setattr_hookimpl_opts
通过@hookimpl方式包装一个func，设置attributes，
set attributes 之后的func 可以在py模块和class类中识别到，
通过PluginManager.register 传入py或者class


import pluggy

hookspec = pluggy.HookspecMarker("myproject")
hookimpl = pluggy.HookimplMarker("myproject")

class MySpec:
    """A hook specification namespace."""

    @hookspec()
    def myhook1(self, arg1, arg2):
        """My special little hook that you can customize."""
    
    @hookspec()
    def myhook2(self, arg1, arg2):
        """My special little hook that you can customize."""

class Plugin_1:
    """A hook implementation namespace."""

    @hookimpl
    def myhook1(self, arg1, arg2):
        print("inside Plugin_1.myhook1()")
        return arg1 + arg2
        
    def myhook2(self, arg1, arg2):
        print("myhook2")
        return arg1 * arg2
        
    myhook2 = hookimpl(function=myhook2)
    
pm = pluggy.PluginManager("myproject")
pm.add_hookspecs(MySpec)
pm.register(Plugin_1())
result = pm.hook.myhook2(arg1=10, arg2=20)
print(result) ## [200]

optionalhook=True

设置True，则未匹配到 hookimpl 相应的 hookspec时，PluginManager.check_pending不会抛出PluginValidationError

note: pluggy 没有严格要求每个 hookimpl 都有相应的 hookspec, 如果希望强制执行此行为，则可以使用 check_pending() 方法运行检查


class MySpec:
    """A hook specification namespace."""

    @hookspec()
    def myhook1(self, arg1, arg2):
        """My special little hook that you can customize."""
    
    @hookspec()
    def myhook2(self, arg1, arg2):
        """My special little hook that you can customize."""

class Plugin_1:
    """A hook implementation namespace."""

    @hookimpl
    def myhook1(self, arg1, arg2):
        print("inside Plugin_1.myhook1()")
        return arg1 + arg2
        
    @hookimpl()
    def myhook2(self, arg1, arg2):
        print("inside Plugin_1.myhook2()")
        return arg1 * arg2
        
pm = pluggy.PluginManager("myproject")
# pm.add_hookspecs(MySpec) # 不添加hookspec
pm.register(Plugin_1()) 
pm.check_pending() # pluggy.manager.PluginValidationError: unknown hook 'myhook1

tryfirst & trylast

tryfirst: 设置为True，则这个hookimpl在 N 个hookimpl队列中越早调用，如果存在多个tryfirst=True的hookimpl,则越晚被register的hookimpl，越早调用

trylast: 设置为True，则这个hookimpl在 N 个hookimpl队列中越迟调用，如果存在多个tryfirst=True的hookimpl,则越早被register的hookimpl，越迟调用

import pluggy

hookspec = pluggy.HookspecMarker("myproject")
hookimpl = pluggy.HookimplMarker("myproject")


class MySpec:
    """A hook specification namespace."""

    @hookspec
    def myhook1(self, arg1, arg2):
        """My special little hook that you can customize."""

    @hookspec
    def myhook2(self, arg1, arg2):
        """My special little hook that you can customize."""

class Plugin_1:
    """A hook implementation namespace."""

    @hookimpl(tryfirst=True)
    def myhook1(self, arg1, arg2):
        return arg1 + arg2

    @hookimpl(trylast=True)
    def myhook2(self, arg1, arg2):
        return arg1 + arg2


class Plugin_2:
    """A 2nd hook implementation namespace."""

    @hookimpl
    def myhook1(self, arg1, arg2):
        return arg1 - arg2

    @hookimpl
    def myhook2(self, arg1, arg2):
        return arg1 - arg2

class Plugin_3:
    """A 2nd hook implementation namespace."""

    @hookimpl(tryfirst=True)
    def myhook1(self, arg1, arg2):
        return arg1 * arg2

    @hookimpl(trylast=True)
    def myhook2(self, arg1, arg2):
        return arg1 * arg2

# create a manager and add the spec
pm = pluggy.PluginManager("myproject")
pm.add_hookspecs(MySpec)
# register plugins
pm.register(Plugin_1())
pm.register(Plugin_2())
pm.register(Plugin_3())
# call our `myhook` hook
results1 = pm.hook.myhook1(arg1=1, arg2=2)
print(results1)
results2 = pm.hook.myhook2(arg1=1, arg2=2)
print(results2)

specname

将 hookimpl 与具有不匹配函数名称的 hookspec 匹配，此时PluginManager.check_pending不会抛出PluginValidationError


import pluggy

hookspec = pluggy.HookspecMarker("myproject")
hookimpl = pluggy.HookimplMarker("myproject")


class MySpec:
    """A hook specification namespace."""

    @hookspec()
    def myhook1(self, arg1, arg2):
        """My special little hook that you can customize."""

    @hookspec()
    def myhook2(self, arg1, arg2):
        """My special little hook that you can customize."""


class Plugin_1:
    """A hook implementation namespace."""

    @hookimpl(specname='myhook1')
    def myhook3(self, arg1, arg2):
        return arg1 + arg2

    @hookimpl(specname='myhook2')
    def myhook4(self, arg1, arg2):
        return arg1 * arg2


pm = pluggy.PluginManager("myproject")
pm.add_hookspecs(MySpec)
pm.register(Plugin_1())
pm.check_pending()  # 不会raise PluginValidationError

hookwrapper

标记hookwrapper=True时，需要被标记的hookimpl的实现为generator(生成器)，即使用了yield，该hookimpl将会包装所有普通的hookimpl,普通的hookimpl调用完成之后，则会调用yield之后的代码
如下outcome 接受了所有被包裹hookimpl的return(pluggy.callers._Result)
如果实现多个hookwrapper，则最后register的优先调用，或者tryfirst=True的优先调用


import pluggy

hookspec = pluggy.HookspecMarker("myproject")
hookimpl = pluggy.HookimplMarker("myproject")


class MySpec:
    """A hook specification namespace."""

    @hookspec()
    def myhook1(self, arg1, arg2):
        """My special little hook that you can customize."""

class Plugin_1:
    """A hook implementation namespace."""

    @hookimpl(specname='myhook1', hookwrapper=True)
    def myhook1(self, arg1, arg2):
        print("Plugin_1")
        outcome = yield
        print("yield Plugin_1", outcome.get_result())


class Plugin_2:
    """A hook implementation namespace."""

    @hookimpl()
    def myhook1(self, arg1, arg2):
        print("Plugin_2")
        return arg1 - arg2


pm = pluggy.PluginManager("myproject")
pm.add_hookspecs(MySpec)
pm.register(Plugin_1())
pm.register(Plugin_2())
# pm.check_pending()  # 不会raise PluginValidationError
result = pm.hook.myhook1(arg1=10, arg2=20)
print(result)

内部是会将pluggy.callers._Result通过send方式传递给hookwrapper内的outcome，以后为pluggy.caller源码部分,gen已经next过一次


if hook_impl.hookwrapper:
    try:
        gen = hook_impl.function(*args)
        next(gen)  # first yield
        teardowns.append(gen)
    except StopIteration:
        _raise_wrapfail(gen, "did not yield")

==================================================
# run all wrapper post-yield blocks
for gen in reversed(teardowns):
    try:
        gen.send(outcome)
        _raise_wrapfail(gen, "has second yield")
    except StopIteration:
        pass