watchdog でファイルの変更を監視する

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)