Edited at

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

More than 1 year has passed since last update.

特定のディレクトリを監視して、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()が呼ばれていました。