🔌

Pluggyとは

2023/02/12に公開

Pluggyとは

Pluggyとは、Pythonにおいてプラガブルなシステムを構築するときに利用できるフレームワークである。
ここでいうプラガブルなシステムとは、プログラム内の動作を都度拡張・変更できるシステムを意図している。

例えば以下のような開発形態で、DataSourceからデータを取ってデータを変換して適当なところへ書き込むような巨大なApplicationを開発していたとする。

このような状態だと、「PluginDeveloper」は「FrameworkDeveloper」の変更の影響を受けやすく、「PluginDeveloper」が本来集中すべきであるPluginの拡張・変更に集中できない。
なのでできれば次のように分離したいよね、ということを実現できるのがここでいうプラガブルなシステム。

特徴

  • 他にもプログラムやライブラリの動作を都度拡張・変更できるものとして、「method overriding」や「monkey patching」というものがあるが、これらは複数の関係者が関わるとなると難しいらしい。Pluggyはより構造化されたアプローチを担っているため、こういった問題を解決できる。
  • pytesttox,devpiのコアフレームワークとなっている。
    • AirflowのListenerの実装にも使われている

構成

主に次のようなComponentによって構成される。

Component About
PluginManager(The Plugin Registry) pluginsの管理
Hookspec(A hook specification) pluginの実装の仕様(hook)について定義
Hookimpl(Implementation) pluginの実装

PluginManager1に対しHookspecがNあったとすると、
HookimplはN(hook数)×N(plugin数)つとなる。

基本的な書き方

import pluggy

# 追加するhookspec,hookimplのdecoratorを定義
hookspec = pluggy.HookspecMarker("myproject")
hookimpl = pluggy.HookimplMarker("myproject")


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


class Plugin_1:
    @hookimpl
    def myhook(self, arg1, arg2):
        print("inside Plugin_1.myhook()")
        return arg1 + arg2
    @hookimpl
    def myhook2(self):
        print("inside Plugin_1.myhook2()")

class Plugin_2:
    @hookimpl
    def myhook(self, arg1, arg2):
        print("inside Plugin_2.myhook()")
        return arg1 - arg2


# PluginManagerの作成
pm = pluggy.PluginManager("myproject")
# hookspecの追加
pm.add_hookspecs(MySpec)
# hookimplの追加
pm.register(Plugin_1())
pm.register(Plugin_2())
# hookimplの削除
## pm.unregister(Plugin_2())
# hook myhookの実行
results = pm.hook.myhook(arg1=1, arg2=2)
print("# [Plugin_1.myhookの結果, Plugin_2.myhookの結果]")
print(results)
# hook myhook2の実行
pm.hook.myhook2(arg1=1)

実行すると次のような結果になる。

inside Plugin_2.myhook()
inside Plugin_1.myhook()
# [Plugin_1.myhookの結果, Plugin_2.myhookの結果]
[-1, 3]
inside Plugin_1.myhook2()

基本的な書き方についての補足

  • HookspecMarker/HookimplMarkerにて渡すproject_nameはPluginManagerにて渡したproject_nameと同一でなければならない。
    • 同一にしないとPluginManagerから検出されないらしい。
  • Hookspecは基本的にDocstringのみ
  • HookspecとHookimplの関係性について
    • Hookspecに定義されていない仕様のHookimplは定義することが基本的できない。
      • 定義されていないHookimplを定義した場合、check_pending()にてPluginValidationErrorが発生する。
  • 任意のHookspecに対する各Hookimplの関係性について(上記の場合だとMySpec.myhookとPlugin_1.myhookなどの関係性について)
    • 関数名は基本的に同じにする必要がある。
    • 関数への引数は「Hookspecの引数>=Hookimplの引数」としなければならない。
  • 実行結果はPluginManagerへのregister()をした順序に対しLIFO(後入先出し)で基本的にListで出力される。

Hookimplについて(詳細)

Hookspecに定義されていない仕様のHookimplを定義することもできる。

「Hookspecに定義されていない仕様のHookimplは定義することが基本的できない」と書いたが、Hookimplへoptionalhook=Trueを指定すれば回避することもできる。

@hookimpl(optionalhook=True)
def not_definition_plugin(self):
    pass
pm.hook.not_definition_plugin()

異なるHookspecにoverrideできる。

Hookimplへspecname=<str>を指定し、異なるHookspecにoverrideできる。

@hookspec(historic=True)
def setup_project(self):
    pass
@hookimpl(specname="setup_project")
def setup_project_override_plugin(self):
    pass
# override元のspecを呼び出すと`setup_project_override_plugin`も呼び出される。
pm.hook.setup_project()

LIFO以外の呼び出し方もできる。

Hookimplへtryfirst=Truetrylast=Trueを指定し、LIFO以外の呼び出しを行うことができる。

class Plugin_1:
    @hookimpl
    def myhook(self):
        print("inside Plugin_1.myhook()")
	return 'plugin_1'

class Plugin_2:
    @hookimpl
    def myhook(self):
        print("inside Plugin_2.myhook()")
	return 'plugin_2'

pm.register(Plugin_1())
pm.register(Plugin_2())
results = pm.hook.myhook(arg1=1, arg2=2)
print(results)

上記の場合、結果は次のようになるわけだが、

inside Plugin_2.myhook()
inside Plugin_1.myhook()
['plugin_2', 'plugin_1']

例えば、Plugin_1.myhookにtryfirst=Trueを付けると、Plugin_1.myhookの実行を一番最初にしてくれる。

・・・
    @hookimpl(tryfirst=True)
    def myhook(self):
・・・

呼び出しの順序だけでなく、結果の順序も変わる。

inside Plugin_1.myhook()
inside Plugin_2.myhook()
['plugin_1', 'plugin_2']

単一のyieldを持つgenerator関数として実装できる。

@contextlib.contextmanagerと同じような動きをさせたいときなどに活用できる。
例えばPlugin_2.myhookの結果とPlugin_3.myhookの結果へint:100を加えたいとする。そのような実装は次のようにかける。

class Plugin_1:
    @hookimpl(hookwrapper=True)
    def myhook(self):
        outcome = yield
        current_result = outcome.get_result()
        print(f"current result is {str(current_result)}")
        replace_result = sum(current_result) + 100
	# 結果の上書き
        outcome.force_result(replace_result)

class Plugin_2:
    @hookimpl
    def myhook(self):
        return 10

class Plugin_3:
    @hookimpl
    def myhook(self):
        return 20

pm.register(Plugin_1())
pm.register(Plugin_2())
pm.register(Plugin_3())
results = pm.hook.myhook()
print(f"replace result is {results}")

実行すると次のような結果が得られる。

current result is [20, 10]
replace result is 130

次の行でoutcomeに入るのは_Result。なのでforce_result()get_result()を使える。

        outcome = yield

Hookspecについて(詳細)

  • 存在する目的

ドキュメント引用
validate each hookimpl ensurring extension writer has correctly defined their callback function implementation.
新たなhookimpl(plugin)を実装する人が実装を正しく定義していることを検証するため

一番最初にregisterされたpluginのみ実行し、結果受け取るように制御することができる。

Hookspecへfirstresult=Trueを指定すれば一番最初にregisterされたpluginのみを実行し、結果を受け取るように制御することができる。
pytest_cmdline_parseで使われていたりする。

class MySpec:
    @hookspec(firstresult=True)
    def myhook2(self):
        """My special little hook that you can customize."""

class Plugin_1:
    @hookimpl
    def myhook2(self):
	print('Plugin_1.myhook2 called')
        return "Plugin_1.myhook2"

class Plugin_2:
    @hookimpl
    def myhook2(self):
        return "Plugin_2.myhook2"

pm.register(Plugin_1())
pm.register(Plugin_2())
results = pm.hook.myhook2()
print(f'myhook2 result is {results}')

例えば上記の場合結果は次のようになる。Plugin_1.myhook2は呼び出しすらされていない。

myhook2 result is Plugin_2.myhook2

ちなみにPlugin_2.myhook2の結果を次のようにNoneにすると

    def myhook2(self):
        return None

Plugin_2.myhook2はpassされ、Plugin_1.myhook2の呼び出しと結果の受け取りが行われる。

Plugin_1.myhook2 called
myhook2 result is Plugin_1.myhook2

PluginManagerについて(詳細)

過去のhookなどもhistoric化できる。

Hookspecへhistoric=Trueを指定すれば過去にあったものとして残しつつということができる。

@hookspec(historic=True)
def myhook(self, arg1, arg2):
    pass

@hookimpl(hookwrapper=True)
def myhook(self, arg1):
    return arg1

上記の状態でpm.hook.myhook()をするとAssertionErrorが発生する。
ただ、過去のhookも呼び出して結果をみたいということもある。その時は当hookに対し、call_historic()で呼び出す。kwargsにはhookへ渡す引数を、result_callbackへは結果が返ってきた時のcallback(func)を渡す。

def callback(result):
    print("historic call result is {result}".format(result=result))

result = pm.hook.myhook.call_historic(kwargs={'arg1':1, 'arg2':2}, result_callback=callback)

実行すると以下のようになり、resultにはNoneが入る。

historic call result is 1

一時的なpluginの実装も対応できる。

hookに対しcall_extra()で一時的なpluginの実装に対応できる。

class Plugin_1:
   @hookimpl
   def myhook(self):
      return 3

pm.register(myhook)
pm.myhook()

例えば上記のようなmyhookへ一時的なplugin(adhoc_funcとadhoc_func2)を対応したいとする。そのような時は次のようにかける。

class Plugin_1:
   @hookimpl
   def myhook(self):
      return 3

def adhoc_func(arg1):
    return arg1
def adhoc_func2(arg2):
    return arg2

pm.register(myhook)
pm.hook.myhook.call_extra([adhoc_func,adhoc_func2], {"arg1": 1, "arg2": 2})

上記の場合呼び出される順序は次のようになる。
Plugin_1.myhook -> adhoc_func2 -> adhoc_func
Plugin_1.myhookへtryfirst=Trueなどを指定してもここにおいては効かないみたい。

そのほか

以下のような機能もある。

  • EntryPointからもpluginをloadすることができる。
pip3 install anysample
pm.load_setuptools_entrypoints("anysample")
  • set_blockedを使って任意のpluginをBlockできる、

まとめ

pluggyの紹介でした。
文章だけでは具体的な実装についてイメージが付かなかった方は公式のeggsampleを試してみると明るくなるかもしれません。
自分はこういった既存の実装を拡張したい時に親クラスを継承し子クラス拡張し呼び出すといったようなやり方(method overriding)を取ることが多いのですが、その場合親クラスをみただけではこれは拡張されるものなのか否かが判断しにくかったり、といったところが課題だなと思ったりしていたので、その辺を解決する手段の一つとしてpluggyは良いものだなと思っています。

参考

Discussion