Pythonで変更のあったモジュールを動的インポート/リロードする

特定のディレクトリを監視して、Pythonモジュールが作成/更新されたらインポート/リロードする例です。

やり方

ファイル監視

ファイル監視は自分で実装できなくもないですが、既存のライブラリを使うと楽に実現できます。

ここではwatchdogというモジュールを使います。
watchdogを使うと特定のディレクトリ内でのファイル作成/更新/削除といったイベント検知ができます。

モジュールの動的インポート

モジュールをインポートするには通常はimport文を使用しますが、モジュール名を文字列で指定することはできません。

インポートしたいモジュールを文字列で動的に指定したい場合には標準モジュールのimpotlib.import_module()を使用します。

import文ではモジュール名を文字列で指定できない

>>> import 'sys'
  File "<stdin>", line 1
    import 'sys'
               ^
SyntaxError: invalid syntax

importlib.import_module()ではモジュール名を文字列で指定する

>>> import importlib
>>> importlib.import_module('sys')
>>> sys = importlib.import_module('sys')
>>> sys
<module 'sys' (built-in)>

モジュールのリロード

インポート済みのモジュールのPythonファイルを更新しても、実行中のプログラムには反映されません。
import文やimpotlib.import_module()を再度実行しても反映されません。

一度インポートしたモジュールを再読み込みするにはimportlib.reload()を使用します。

>>> with open('a.py', 'w') as f:
...     f.write('def f():\n    print("1")')
...
23
>>> import a
>>> a.f()
1
>>> with open('a.py', 'w') as f:
...     f.write('def f():\n    print("2")')
...
23
>>> a.f()
1
>>> import a
>>> a.f()
1
>>> import importlib
>>> a = importlib.import_module('a')
>>> a.f()
1
>>> a = importlib.reload(a)
>>> a.f()
2

サンプルコード

Python3.4以上で動作します。

plugin_manager.py

import sys
import time
from importlib import import_module, reload
from pathlib import Path

from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
from watchdog.observers import Observer


class PluginManager:

    class Handler(PatternMatchingEventHandler):

        def __init__(self, manager: 'PluginManager', *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.manager = manager

        def on_created(self, event: FileSystemEvent):
            print(event)
            if event.src_path.endswith('.py'):
                self.manager.load_plugin(Path(event.src_path))

        def on_modified(self, event):
            print(event)

    def __init__(self, path: str):
        self.plugins = {}
        self.path = path
        self.observer = Observer()

        sys.path.append(self.path)

    def start(self):

        self.scan_plugin()

        self.observer.schedule(self.Handler(self, patterns='*.py'), self.path)
        self.observer.start()

    def stop(self):
        self.observer.stop()
        self.observer.join()

    def scan_plugin(self):
        for file_path in Path(self.path).glob('*.py'):
            self.load_plugin(file_path)

    def load_plugin(self, file_path):
        module_name = file_path.stem
        if module_name not in self.plugins:
            self.plugins[module_name] = import_module(module_name)
            print('{} loaded.'.format(module_name))
        else:
            self.plugins[module_name] = reload(self.plugins[module_name])
            print('{} reloaded.'.format(module_name))


def main():

    plugin_manager = PluginManager('plugins/')
    plugin_manager.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        plugin_manager.stop()

if __name__ == '__main__':
    main()

実行結果

plugins/にa.pyがある状態で起動

a loaded.

plugins/にb.pyを作成

<FileCreatedEvent: src_path='plugins/b.py'>
b loaded.
<DirModifiedEvent: src_path='plugins/'>
<DirModifiedEvent: src_path='plugins/__pycache__'>

b.pyに`print('bbb')`と書いて保存

<FileCreatedEvent: src_path='plugins/b.py'>
bbb
b reloaded.
<DirModifiedEvent: src_path='plugins/'>
<DirModifiedEvent: src_path='plugins/__pycache__'>

補足

• watchdogのハンドラではファイル更新を検知するとon_modified()が呼ばれるはずだと思うんですが、on_created()が呼ばれていました。