現代のシステム開発やデータ処理の現場において、特定のフォルダをリアルタイムで監視し、ファイルの変化をトリガーにして自動的に処理を実行する仕組みは非常に重要な役割を担っています。
例えば、画像がアップロードされた瞬間にリサイズ処理を開始したり、ログファイルが更新された際に特定のキーワードを抽出して通知を送ったりといった自動化が挙げられます。
Pythonでこのようなファイルシステムの監視を実現するために、最も広く普及し、信頼性が高いライブラリがwatchdogです。
OS固有のイベント通知機能を抽象化し、クロスプラットフォームで動作するこのライブラリは、複雑なポーリング処理を自作することなく、最小限のコードで高度な自動化を実現してくれます。
本記事では、watchdogの基本的な導入方法から、実務で役立つ具体的な自動化の実装パターンまでを詳しく解説します。
Pythonにおけるフォルダ監視の重要性
自動化のワークフローを構築する際、プログラムを一定時間おきに実行してフォルダの中身を確認する「ポーリング方式」は、リソースの無駄遣いや反応の遅延という課題を抱えています。
これに対し、ファイルシステムのイベントを直接検知する方式を採用すれば、変更があった瞬間にだけプログラムを動かすことが可能になります。
Pythonのwatchdogライブラリは、Windows、macOS、Linuxといった主要なOSが提供するファイルイベント通知API（inotify、FSEvents、ReadDirectoryChangesWなど）をラップしています。
これにより、開発者はOSの違いを意識することなく、「ファイルが作られた」「削除された」「書き換えられた」といったイベントに対して一貫したコードを記述できるようになります。
watchdogライブラリのインストールと基本構成
まずは、開発環境にwatchdogをインストールしましょう。
Pythonがインストールされている環境であれば、標準的なパッケージ管理ツールであるpipを使用して簡単に導入できます。
pip install watchdogwatchdogは大きく分けて2つの主要なコンポーネントで構成されています。
- Observer（オブザーバー）: フォルダの変更を監視するバックグラウンドスレッドです。
- EventHandler（イベントハンドラ）: イベント（作成、修正、削除、移動）が発生した際に、どのようなアクションを実行するかを定義するクラスです。
この2つを組み合わせることで、監視対象のパスで何かが起きたときに特定の関数を呼び出す仕組みを構築します。
フォルダ監視の基本実装手順
最もシンプルなフォルダ監視のスクリプトを作成してみましょう。
以下のコードは、指定したフォルダ内でのファイル操作をすべてコンソールに出力する基本的なテンプレートです。
import time
import sys
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SimpleHandler(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}")
if __name__ == "__main__":
# 監視対象のパス（現在のディレクトリを対象にする場合）
path = "."
# ハンドラとオブザーバーの設定
event_handler = SimpleHandler()
observer = Observer()
# 監視のスケジュール登録（recursive=Trueでサブフォルダも監視）
observer.schedule(event_handler, path, recursive=True)
# 監視開始
print(f"監視を開始しました: {path}")
observer.start()
try:
while True:
# CPU負荷を下げるための待機
time.sleep(1)
except KeyboardInterrupt:
# Ctrl+Cで停止
observer.stop()
# スレッドの終了を待機
observer.join()このスクリプトを実行した状態で、監視対象フォルダに「test.txt」というファイルを作成し、その後中身を書き換えて削除すると、以下のようなログが出力されます。
監視を開始しました: .
作成されました: ./test.txt
変更されました: ./test.txt
削除されました: ./test.txtこのように、FileSystemEventHandlerを継承したクラスを作成し、各種メソッド（on_createdなど）をオーバーライドするだけで、イベントごとの処理を記述できるのがwatchdogの大きな特徴です。
主要なイベントハンドラの種類と使い分け
watchdogには、用途に応じた複数のハンドラクラスが用意されています。
これらを使い分けることで、より効率的な監視が可能になります。
| クラス名 | 特徴と用途 |
|---|---|
FileSystemEventHandler | 基本的なハンドラ。すべてのイベントを自分で定義したい場合に使用します。 |
PatternMatchingEventHandler | 特定のファイルパターン（拡張子など）のみを対象にしたい場合に便利です。 |
RegexMatchingEventHandler | 正規表現を用いて、より複雑なファイル名ルールで監視対象を絞り込みたい場合に使用します。 |
LoggingEventHandler | デバッグ用。発生したすべてのイベントをPythonのloggingモジュール経由で出力します。 |
実務で特によく使われるのは、PatternMatchingEventHandlerです。
例えば、CSVファイルが追加されたときだけ処理を行いたいといったケースでは、このハンドラを使うことで、不要なイベント処理をフィルタリングし、コードを簡潔に保つことができます。
実践：特定の拡張子を監視して自動処理する手法
次に、実用的な例として「特定のフォルダに画像（.pngや.jpg）が保存されたら、自動的に別のフォルダへ移動させる」という自動整理ツールを作成してみましょう。
import os
import shutil
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler
class ImageOrganizerHandler(PatternMatchingEventHandler):
"""
特定の画像ファイルのみを監視し、自動で移動させるハンドラ
"""
def __init__(self):
# 監視対象にするファイルパターンを指定
super().__init__(patterns=["*.png", "*.jpg", "*.jpeg"],
ignore_directories=True)
def on_created(self, event):
# 新しい画像が見つかった時の処理
file_path = event.src_path
file_name = os.path.basename(file_path)
destination_dir = "./organized_images"
# 保存先フォルダが存在しない場合は作成
if not os.path.exists(destination_dir):
os.makedirs(destination_dir)
# ファイル移動の実行
dest_path = os.path.join(destination_dir, file_name)
# ファイルが書き込み中の場合を考慮して少し待機（実用上の工夫）
import time
time.sleep(0.5)
try:
shutil.move(file_path, dest_path)
print(f"移動完了: {file_name} -> {destination_dir}")
except Exception as e:
print(f"エラーが発生しました: {e}")
if __name__ == "__main__":
path = "./uploads" # 監視対象フォルダ
if not os.path.exists(path):
os.makedirs(path)
event_handler = ImageOrganizerHandler()
observer = Observer()
observer.schedule(event_handler, path, recursive=False)
print(f"{path} 内の画像を監視中...")
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()この実装のポイントは、patterns引数を使用して監視対象を画像ファイルに限定している点です。
これにより、一時ファイルや設定ファイルなど、関係のないファイルが作成されてもプログラムは反応せず、必要な時だけ動作する効率的なシステムになります。
効率的な自動化のための高度なテクニック
watchdogを実務レベルのシステムに組み込む際には、いくつか注意すべきテクニックがあります。
1. 二重イベントの回避
一部のアプリケーション（特にテキストエディタやIDE）は、ファイルを保存する際に「一時ファイルの作成」「上書き」「削除」といった複数の工程を高速に行います。
そのため、一度の保存に対してon_modifiedが複数回発生することがあります。
これを回避するためには、最後にイベントが発生してからの経過時間をチェックするデバウンス（防振）処理を入れるか、ハッシュ値を用いてファイル内容が実際に変わったかを確認する処理を加えるのが効果的です。
2. サブフォルダ監視のパフォーマンス
observer.scheduleの引数でrecursive=Trueを設定すると、配下のすべてのサブフォルダが監視対象になります。
非常に深いディレクトリ構造や、数万個のファイルが含まれるフォルダを監視する場合、OSのファイル記述子の制限に達したり、CPU負荷が高まったりする可能性があります。
大規模なディレクトリを監視する場合は、監視対象を最小限に絞るか、無視したいフォルダをignore_patternsで指定することを検討してください。
3. ファイルの書き込み完了を待つ
ファイルが作成された直後（on_created発生時）は、まだ大きなファイルの場合、書き込みが完了していないことがあります。
この状態でファイルを開こうとするとエラーになるため、ファイルサイズが一定時間変わらなくなるのを待つ、あるいはリトライ処理を入れるといった工夫が重要です。
watchdogを活用した自動化のアイデア集
このライブラリを活用することで、以下のようなワークフローが簡単に構築できます。
データの自動バックアップ
特定の作業フォルダでファイルが更新されるたびに、日付スタンプを付けてバックアップサーバやクラウドストレージへコピーを送信します。
shutilやboto3（AWS SDK）と組み合わせることで強力なツールになります。
AIモデルの自動推論
画像や音声ファイルが特定のフォルダに置かれたことを検知し、即座にAI推論モデルに読み込ませて結果を出力します。
バッチ処理を待つ必要がなく、ほぼリアルタイムの解析システムを構築可能です。
開発環境のホットリロード
設定ファイル（YAMLやJSON）が更新された際に、実行中のアプリケーションの設定を動的に読み込み直す仕組みも構築できます。
サーバーを再起動することなく設定を反映させられるため、開発効率が劇的に向上します。
まとめ
Pythonのwatchdogライブラリは、シンプルながらも非常に強力なファイルシステム監視ツールです。
OSレベルのイベントを効率的にキャッチし、Pythonの柔軟な記述力と組み合わせることで、日常の煩雑なファイル操作を驚くほど簡単に自動化できます。
本記事で解説したObserverとEventHandlerの基本構造を理解すれば、あとは自身の目的に合わせてハンドラ内の処理をカスタマイズするだけです。
「手動で行っているファイル操作」を「イベント駆動の自動処理」に置き換えることで、ヒューマンエラーを減らし、より付加価値の高い業務に時間を割けるようになるでしょう。
まずは小さなフォルダの監視から始めて、Pythonによる効率的な自動化の世界を体験してみてください。