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
