fumbling around with pluggy
ยค
https://pluggy.readthedocs.io/
- register/unregister plugins
- define specs and implementations at the same time.
import pluggy
we're going to think about a sample PROG
ram that has an interface.
PROG = "sample-program"
the specification
decorates the functions and signatures of our interface.
the implementation
uses the specification
for consistency when it actually defines computational work.
implementation, specification = pluggy.HookimplMarker(PROG), pluggy.HookspecMarker(PROG)
PROG
s specification
for my_plugin
provides a default interface.
@implementation
@specification(firstresult=False)
def my_plugin(a, b, c): return "".join(map(str, (a,b,c)))
we register the specifaction
onto a plugin manager
manager = pluggy.PluginManager(PROG)
our specification
lives in the import __main__
or MAIN
namespace.
manager.add_hookspecs(__main__ := __import__(__name__))
assert "my_plugin" in dir(manager.hook), "the plugin is registered"
now we can add our implementation
of my_plugin
in the __main__
module
manager.register(__main__);
now we can execute our manager.hook.my_plugin
method
manager.hook.my_plugin(a=10, b=20, c=30)
let's add another implementation. we can't name it my_plugin
otherwise we'll lose the scope of our previous method. here we name our method whatever we want and explicitly define the implementation:specname
it refers to.
@implementation(specname="my_plugin")
def another_plugin(a, b): return (a, b)
now seems like a good time test what async
functions do
@implementation(specname="my_plugin", trylast=True)
async def async_my_plugin(a, b): return (a, b)
reregistering the __main__
gives an error because pluggy
doesn't allow someone register the module twice
try: manager.register(__main__); assert False, "can't register the module twice"
except ValueError: assert True
the proper registration requires we unregister the module first.
manager.unregister(__main__)
manager.register(__main__);
now our invocation finds both implementations.
(results := manager.hook.my_plugin(a=10, b=20, c=30))
async_my_plugin
evaluates to a coroutine that we'd have to handle properly.
assert __import__("inspect").iscoroutine(results[-1])