A wrapper of "pluggy" to support asyncio and context managers


Keywords
asyncio, nextline, pluggy
License
MIT
Install
pip install apluggy==1.0.2

Documentation

apluggy

PyPI - Version PyPI - Python Version

Test Status Test Status codecov

A wrapper of pluggy to support asyncio and context managers.

This package provides a subclass of pluggy.PluginManager which

  • allows async functions, context managers, and async context managers to be hooks
  • and accepts plugin factories in addition to plugin instances for registration.

Table of Contents


Installation

You can install apluggy with pip:

pip install apluggy

How to use

Here, we show a simple example of how to use apluggy.

We only describe the usage of additional features provided by apluggy. For the usage of pluggy itself, please refer to the pluggy documentation.

Start Python

You can try this example in a Python interpreter.

$ python
Python 3.10.13 (...)
...
...
>>>

Import packages

Import necessary packages of this example.

>>> import asyncio
>>> import apluggy as pluggy
>>> from apluggy import asynccontextmanager, contextmanager

In this example, apluggy is imported with the alias pluggy.

The decorators asynccontextmanager and contextmanager are imported from apluggy. They are wrappers of the decorators of the same names in the contextlib package. The wrappers preserve the signatures of decorated functions, which are necessary for pluggy to pass arguments to hook implementations correctly. (The decorator contextmanger in apluggy is the same object as the decorator contextmanager in the decorator package. The decorator package does not provide asynccontextmanager decorator as of version 5.1. The decorator asynccontextmanger in apluggy is implemented in a similar way as the decorator contextmanager in the decorator package.)

Create hook specification and implementation decorators

>>> hookspec = pluggy.HookspecMarker('project')
>>> hookimpl = pluggy.HookimplMarker('project')

Define hook specifications

In this example, we define three hooks: async function, context manager, and async context manager.

>>> class Spec:
...     """A hook specification namespace."""
...
...     @hookspec
...     async def afunc(self, arg1, arg2):
...         pass
...
...     @hookspec
...     @contextmanager
...     def context(self, arg1, arg2):
...         pass
...
...     @hookspec
...     @asynccontextmanager
...     async def acontext(self, arg1, arg2):
...         pass

Define plugins

We define two plugins as classes. Each plugin implements the three hooks defined above.

>>> class Plugin_1:
...     """A hook implementation namespace."""
...
...     @hookimpl
...     async def afunc(self, arg1, arg2):
...         print('inside Plugin_1.afunc()')
...         return arg1 + arg2
...
...     @hookimpl
...     @contextmanager
...     def context(self, arg1, arg2):
...         print('inside Plugin_1.context(): before')
...         yield arg1 + arg2
...         print('inside Plugin_1.context(): after')
...
...     @hookimpl
...     @asynccontextmanager
...     async def acontext(self, arg1, arg2):
...         print('inside Plugin_1.acontext(): before')
...         yield arg1 + arg2
...         print('inside Plugin_1.acontext(): after')

>>> class Plugin_2:
...     """A 2nd hook implementation namespace."""
...
...     @hookimpl
...     async def afunc(self, arg1, arg2):
...         print('inside Plugin_2.afunc()')
...         return arg1 - arg2
...
...     @hookimpl
...     @contextmanager
...     def context(self, arg1, arg2):
...         print('inside Plugin_2.context(): before')
...         yield arg1 - arg2
...         print('inside Plugin_2.context(): after')
...
...     @hookimpl
...     @asynccontextmanager
...     async def acontext(self, arg1, arg2):
...         print('inside Plugin_2.acontext(): before')
...         yield arg1 - arg2
...         print('inside Plugin_2.acontext(): after')

Create a plugin manager and register plugins

Plugins can be registered as instances or factories. In the following example, we register two plugins: Plugin_1 as an instance, and Plugin_2 as a factory.

>>> pm = pluggy.PluginManager('project')
>>> pm.add_hookspecs(Spec)
>>> _ = pm.register(Plugin_1())  # instantiation is optional.
>>> _ = pm.register(Plugin_2)  # callable is considered a plugin factory.

Pluggy accepts a class or module as a plugin. However, it actually accepts a class instance, not a class itself. Consequently, when plugins are loaded with load_setuptools_entrypoints(), the entry points must be class instances or modules. Classes themselves cannot be used as entry points (if understood correctly).

So that classes themselves can be entry points, apluggy accepts a class itself for a plugin registration. When apluggy receives a callable object, apluggy considers the object as a plugin factory.

Call hooks

The following example shows how to call hooks.

Async function

>>> async def call_afunc():
...     results = await pm.ahook.afunc(arg1=1, arg2=2)  # ahook instead of hook
...     print(results)

>>> asyncio.run(call_afunc())
inside Plugin_2.afunc()
inside Plugin_1.afunc()
[-1, 3]

Context manager

>>> with pm.with_.context(arg1=1, arg2=2) as y:  # with_ instead of hook
...     print(y)
inside Plugin_2.context(): before
inside Plugin_1.context(): before
[-1, 3]
inside Plugin_1.context(): after
inside Plugin_2.context(): after

In the reverse order:

>>> with pm.with_reverse.context(arg1=1, arg2=2) as y:  # with_reverse instead of hook
...     print(y)
inside Plugin_1.context(): before
inside Plugin_2.context(): before
[3, -1]
inside Plugin_2.context(): after
inside Plugin_1.context(): after

Async context manager

>>> async def call_acontext():
...     async with pm.awith.acontext(arg1=1, arg2=2) as y:  # awith instead of hook
...         print(y)

>>> asyncio.run(call_acontext())
inside Plugin_2.acontext(): before
inside Plugin_1.acontext(): before
[-1, 3]
inside Plugin_1.acontext(): after
inside Plugin_2.acontext(): after

In the reverse order:

>>> async def call_acontext():
...     async with pm.awith_reverse.acontext(arg1=1, arg2=2) as y:  # awith_reverse instead of hook
...         print(y)

>>> asyncio.run(call_acontext())
inside Plugin_1.acontext(): before
inside Plugin_2.acontext(): before
[3, -1]
inside Plugin_2.acontext(): after
inside Plugin_1.acontext(): after

Links


License

  • apluggy is licensed under the MIT license.