skip to main content

@tonyfast s notebooks

site navigation
notebook summary
title
fumbling around with pluggy
description
https://pluggy.readthedocs.io/
cells
28 total
14 code
state
executed in order
kernel
Python [conda env:root] *
language
python
name
conda-root-py
lines of code
39
outputs
2
table of contents
{"kernelspec": {"display_name": "Python [conda env:root] *", "language": "python", "name": "conda-root-py"}, "language_info": {"codemirror_mode": {"name": "ipython", "version": 3}, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.9.10"}, "title": "fumbling around with pluggy", "description": "https://pluggy.readthedocs.io/"}
notebook toolbar
Activate
cell ordering
1

fumbling around with

https://pluggy.readthedocs.io/

  • register/unregister plugins
  • define specs and implementations at the same time.
2
    import pluggy
3

we're going to think about a sample PROG ram that has an interface.

4
    PROG = "sample-program"
5

the specification decorates the functions and signatures of our interface. the implementation uses the specification for consistency when it actually defines computational work.

6
    implementation, specification = pluggy.HookimplMarker(PROG), pluggy.HookspecMarker(PROG)
7

PROG s specification for my_plugin provides a default interface.

8
    @implementation
    @specification(firstresult=False)
    def my_plugin(a, b, c): return "".join(map(str, (a,b,c)))
9

we register the specifaction onto a plugin manager

10
    manager = pluggy.PluginManager(PROG)
11

our specification lives in the import __main__ or MAIN namespace.

12
    manager.add_hookspecs(__main__ := __import__(__name__))
    assert "my_plugin" in dir(manager.hook), "the plugin is registered"
13

now we can add our implementation of my_plugin in the __main__ module

14
    manager.register(__main__);
15

now we can execute our manager.hook.my_plugin method

16
    manager.hook.my_plugin(a=10, b=20, c=30)
1 outputs.
['102030']
17

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.

18
    @implementation(specname="my_plugin")
    def another_plugin(a, b): return (a, b)    
19

now seems like a good time test what async functions do

20
    @implementation(specname="my_plugin", trylast=True)
    async def async_my_plugin(a, b): return (a, b)    
21

reregistering the __main__ gives an error because pluggy doesn't allow someone register the module twice

22
    try: manager.register(__main__); assert False, "can't register the module twice"
    except ValueError: assert True
23

the proper registration requires we unregister the module first.

24
    manager.unregister(__main__)
    manager.register(__main__);
25

now our invocation finds both implementations.

26
    (results := manager.hook.my_plugin(a=10, b=20, c=30))
1 outputs.
['102030', (10, 20), <coroutine object async_my_plugin at 0x7feab0168040>]
27

async_my_plugin evaluates to a coroutine that we'd have to handle properly.

28
    assert __import__("inspect").iscoroutine(results[-1])