watchdog ライブラリを使うと、ファイルやディレクトリの変更をリアルタイムで監視できる。ログ監視、自動ビルド、同期処理などに便利だ。
インストール
pip install watchdog
基本的な使い方
ファイル変更を検知するには、イベントハンドラを定義して Observer に登録する。
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class MyHandler(FileSystemEventHandler):
def on_created(self, event):
print(f'作成: {event.src_path}')
def on_modified(self, event):
print(f'変更: {event.src_path}')
def on_deleted(self, event):
print(f'削除: {event.src_path}')
def on_moved(self, event):
print(f'移動: {event.src_path} -> {event.dest_path}')
# 監視を開始
observer = Observer()
observer.schedule(MyHandler(), path='./watch_dir', recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
recursive=True を指定すると、サブディレクトリも監視対象になる。
イベントの種類
watchdog は以下のイベントを検知する。
| イベント | 説明 |
|---|---|
| on_created | ファイルまたはディレクトリが作成された |
| on_modified | ファイルまたはディレクトリが変更された |
| on_deleted | ファイルまたはディレクトリが削除された |
| on_moved | ファイルまたはディレクトリが移動/リネームされた |
特定のファイルだけを監視する
パターンマッチングでフィルタリングする場合は PatternMatchingEventHandler を使う。
from watchdog.events import PatternMatchingEventHandler
class MyHandler(PatternMatchingEventHandler):
def __init__(self):
super().__init__(
patterns=['*.py', '*.json'],
ignore_patterns=['*.pyc', '__pycache__/*'],
ignore_directories=True
)
def on_modified(self, event):
print(f'Python/JSON ファイルが変更: {event.src_path}')
ディレクトリのみを監視する
event.is_directory でファイルとディレクトリを区別できる。
class DirOnlyHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
print(f'ディレクトリ作成: {event.src_path}')
実用的な例：ファイル変更時に自動リロード
設定ファイルの変更を検知して自動でリロードする例を示す。
import json
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
config = {}
def load_config():
global config
with open('config.json') as f:
config = json.load(f)
print(f'設定を読み込みました: {config}')
class ConfigHandler(FileSystemEventHandler):
def on_modified(self, event):
if event.src_path.endswith('config.json'):
load_config()
# 初期読み込み
load_config()
observer = Observer()
observer.schedule(ConfigHandler(), path='.', recursive=False)
observer.start()
イベントのデバウンス
ファイル保存時に複数のイベントが発火することがある。デバウンス処理を入れると、短時間の連続イベントをまとめられる。
import time
from threading import Timer
class DebouncedHandler(FileSystemEventHandler):
def __init__(self, callback, delay=0.5):
self.callback = callback
self.delay = delay
self.timer = None
def on_modified(self, event):
if self.timer:
self.timer.cancel()
self.timer = Timer(self.delay, self.callback, [event])
self.timer.start()
def handle_change(event):
print(f'処理: {event.src_path}')
handler = DebouncedHandler(handle_change, delay=0.5)
コンテキストマネージャで使う
クリーンアップを自動化するには、コンテキストマネージャでラップする。
from contextlib import contextmanager
@contextmanager
def watch_directory(path, handler, recursive=True):
observer = Observer()
observer.schedule(handler, path=path, recursive=recursive)
observer.start()
try:
yield observer
finally:
observer.stop()
observer.join()
# 使用例
with watch_directory('./watch_dir', MyHandler()) as observer:
while True:
time.sleep(1)
