【Python】Watchdogを使ってフォルダ監視をする方法

Pythonでフォルダ監視をする際に使えるWatchdogモジュールの使い方について解説します。

Contents

1 Watchdogによるフォルダ監視
- 1.1 Watchdogのインストール方法
- 1.2 公式ドキュメントのサンプルプログラム
2 Watchdogを使って独自の監視クラスを作成する方法
- 2.1 プログラムの詳細説明

Watchdogによるフォルダ監視

サーバーやPCのフォルダを監視して、フォルダに変化があった場合にプログラムで処理をするといったことをしたい場合があります。

例えばですが、あるフォルダを監視していてファイルがそのフォルダにコピーされたら「データベースへ書き込む」「バックアップを取る」といったようなことがあげられるかと思います。

このような場合、フォルダの変化を監視することが必要になってくるわけですが、Pythonで実現する場合にはWatchdogモジュールを使用することができます。

本記事では、Watchdogモジュールの簡単な使い方を公式ドキュメントにあるQuickStartのサンプルプログラムで紹介するとともに、Watchdogのクラスを継承して独自監視クラスを作成する例について紹介します。

Note

Watchdogの公式ドキュメントページはこちらを参考にしてください。

Watchdogのインストール方法

Watchdogモジュールは以下のようにpipを使ってインストールします。

pip install watchdog

インストールが完了すればWatchdogモジュールを使う準備が整います。

公式ドキュメントのサンプルプログラム

Watchdogの公式ドキュメントにはQuickstartというページがあり、Watchdogを使用するためのサンプルソースコードが紹介されています。このサンプルソースコードでWatchdogの基本的な使用方法が分かります。

【※以下ソースコードは公式ドキュメントのQuickstartより引用】

import logging
import sys
import time

from watchdog.events import LoggingEventHandler
from watchdog.observers import Observer

if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    path = sys.argv[1] if len(sys.argv) > 1 else "."
    event_handler = LoggingEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

【実行結果例】
----- ①
2022-04-23 08:29:42 - Created directory: .\新しいフォルダー
2022-04-23 08:29:43 - Modified directory: .\新しいフォルダー
----- ②
2022-04-23 08:30:07 - Moved directory: from .\新しいフォルダー to .\test
2022-04-23 08:30:07 - Modified directory: .\test
----- ③
2022-04-23 08:30:21 - Created file: .\test\sample.txt
2022-04-23 08:30:21 - Modified directory: .\test
2022-04-23 08:30:21 - Modified file: .\test\sample.txt
----- ④
2022-04-23 08:30:28 - Deleted file: .\test\sample.txt
2022-04-23 08:30:28 - Modified directory: .\test

上記実行結果例については、以下の手順で監視対象フォルダを操作をした場合の結果です。なお「—– ①」のように書いている部分は分かりやすいようにどの番号の操作か後で補足したものであり、実際出力される文字列ではありませんのでご注意ください。

【操作手順】

「新しいフォルダー」を作成する。
上記１で作成したフォルダの名称を「test」に変更する。
testフォルダの下に「sample.txt」をコピーして配置する。
sample.txtを削除する。

【サンプルソースコード説明】

サンプルソースコードの各部分について説明をしていきます。Watchdogを使用するためのimport部分が以下の部分です。

from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

監視を実行するためのクラスが「Observer」で、フォルダの変更がされた時にどのような動作をするかのハンドラーが「LoggingEventHandler」です。この例は、フォルダを監視し、変化をロギングするサンプルプログラムになっています。

メインの中の以下部分は、loggingモジュールのログ設定と引数処理をしているところで、詳細は割愛しますがbasicConfigでログレベルやフォーマットを設定し、コマンドライン引数で監視するフォルダパスが設定された場合はpathに設定します。なお、引数で指定がない場合は’.’ということで実行フォルダが対象となります。

loggingについて興味がある方は「loggingの基本的な使い方」でもまとめていますので興味があれば参考にしてください。

   logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    path = sys.argv[1] if len(sys.argv) > 1 else "."

以下の部分が、実際の監視に関連する部分です。

    event_handler = LoggingEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

まず、event_handlerとしてLoggingEventHandlerを設定しています。その後、監視のためのObserverをインスタンス化し、scheduleメソッドでハンドラーと監視対象パスを指定します。recursiveはフォルダ階層をたどって監視するか否かを指定するもので、Trueとした場合は、サブフォルダの変更も監視対象となります。

observer.start以降が監視を動作させるためのもので、この部分は定型と理解してしまってよいかなと思います。無限ループで動作し、Ctrl+CといったKeyboardInterruptが発生した場合に監視を終了します。

実行結果を改めて見てみると、1操作に1イベントというわけではなく、複数のイベントが発生しているのが分かるかと思います。例えば、ファイルを削除した場合には、ファイルを削除したというイベントと、ファイルが配置されていたフォルダが変更されたというイベントが発生しています。

以上が、QuickStartのサンプルソースコードの概要説明です。以降では、Watchdogを使って独自の監視クラスを作成する方法について説明します。

Watchdogを使って独自の監視クラスを作成する方法

Watchdogの使い方として、公式ドキュメントのQuickstartにあるサンプルソースコードの使い方を見てきました。フォルダを監視して独自処理を実装したい場合には、「FileSystemEventHandler」というクラスを継承して、メソッドをオーバーライドすることで独自処理を追加することができます。

なお、Quickstartで見てきたLoggingEventHandler自体もFileSystemHandlerを継承しているものになっています。

FileSystemEventHandlerクラスを継承して使用するサンプルプログラムを以下に示します。以降の節で具体的な内容を説明していきたいと思います。

import time
from argparse import ArgumentParser

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer


class MyWatchHandler(FileSystemEventHandler):
    """監視ハンドラ"""

    def __init__(self):
        """コンストラクタ"""
        super().__init__()

    def on_any_event(self, event):
        """全イベント検知

        Args:
            event: 検知イベント
        """
        # print(f'[on_any_event] {event}')
        pass

    def on_moved(self, event):
        """moved検知関数

        Args:
            event: イベント
        """
        print(f"[on_moved] {event}")

    def on_created(self, event):
        """created検知関数

        Args:
            event: イベント
        """
        print(f"[on_created] {event}")

    def on_modified(self, event):
        """modified検知関数

        Args:
            event: イベント
        """
        print(f"[on_modified] {event}")

    def on_deleted(self, event):
        """deleted検知関数

        Args:
            event: イベント
        """
        print(f"[on_deleted] {event}")


def monitor(path):
    """監視実行関数

    Args:
        path: 監視対象パス
    """
    event_handler = MyWatchHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()


def main():
    """メイン関数"""
    # ===== ArgumentParserの設定
    parser = ArgumentParser(description="Monitoring Tool")
    # 引数の処理
    parser.add_argument("-p", "--path", action="store", dest="path", help="監視対象パス")
    # コマンドライン引数のパース
    args = parser.parse_args()
    # 引数の取得
    path = args.path
    # pathの指定がない場合は実行ディレクトリに設定
    if path is None:
        path = "."

    # モニター実行
    monitor(path)


if __name__ == "__main__":
    main()

【実行結果例】
----- ①
[on_created] <DirCreatedEvent: event_type=created, src_path='.\\新しいフォルダー', is_directory=True>
[on_modified] <DirModifiedEvent: event_type=modified, src_path='.\\新しいフォルダー', is_directory=True>
----- ②
[on_moved] <DirMovedEvent: src_path='.\\新しいフォルダー', dest_path='.\\test', is_directory=True>
[on_modified] <DirModifiedEvent: event_type=modified, src_path='.\\test', is_directory=True>
----- ③
[on_created] <FileCreatedEvent: event_type=created, src_path='.\\test\\sample.txt', is_directory=False>
[on_modified] <DirModifiedEvent: event_type=modified, src_path='.\\test', is_directory=True>
[on_modified] <FileModifiedEvent: event_type=modified, src_path='.\\test\\sample.txt', is_directory=False>
----- ④
[on_deleted] <FileDeletedEvent: event_type=deleted, src_path='.\\test\\sample.txt', is_directory=False>
[on_modified] <DirModifiedEvent: event_type=modified, src_path='.\\test', is_directory=True>

上記実行結果例については、以下の手順でフォルダ監視対象操作をした場合の結果です。なお「—– ①」のように書いている部分は分かりやすいようにどの番号の操作か後で補足したものであり、実際出力される文字列ではありませんのでご注意ください。

【操作手順】

「新しいフォルダー」を作成する。
上記１で作成したフォルダの名称を「test」に変更する。
testフォルダの下に「sample.txt」をコピーして配置する。
sample.txtを削除する。

プログラムの詳細説明

上記で紹介したサンプルプログラムの内容を説明します。プログラム全体の構成としては以下のようになっています。

(クラス) MyWatchHandler：独自の監視クラス定義
(関数) monitor：監視実行用関数
(関数) main：メイン処理（引数処理とmonitorの呼び出し）

MyWatchHandler：独自の監視クラス定義

MyWatchHandlerクラスの定義部分で、FileSystemEventHandlerクラスを継承して独自のクラスを定義しています。クラス名は適当なのでもちろん任意に変えていただいて大丈夫です。

コンストラクタでは、以下の部分でFileSystemEventHandlerの__init__を呼び出しています。

    def __init__(self):
        """コンストラクタ"""
        super().__init__()

その後、各種メソッドをオーバーライドしています。オーバーライドしているメソッドは以下の通りです。

メソッド	検知する操作
on_any_event	全イベント検知
on_moved	名称変更
on_created	作成
on_modified	変更
on_deleted	削除

今回のサンプルプログラムは各関数で発生するイベントを確認するために以下のように内容を表示するだけというシンプルなものにしています。オーバーライドしているメソッドの構成は同じため、on_movedメソッドのみ示します。

    def on_moved(self, event):
        """moved検知関数

        Args:
            event: イベント
        """
        print(f"[on_moved] {event}")

実行結果を見てもらえば分かりますが、eventに設定されるのは「DirCreatedEvent」「DirModifiedEvent」「FileCreatedEvent」…といったクラスのインスタンスになります。対象がディレクトリかファイルか、どういった操作かによって変わってきます。

なお、on_any_eventはすべてのイベントで動作するので結果表示が多くなってしまうため、以下のようにpassとしています。確認したい場合は、printしているコメントアウト部分を外して確認してみてください。

    def on_any_event(self, event):
        """全イベント検知

        Args:
            event: 検知イベント
        """
        # print(f'[on_any_event] {event}')
        pass

上記例は、独自監視クラスを作成するための構成を示す目的で、printするだけのメソッドですが、実際には検知したイベントに対して、独自の実装（例えばファイルをデータベースに書き込む、バックアップするなど）を加えていくことができます。

Note

on_movedメソッドは、API仕様上「Called when a file or a directory is moved or renamed.」となっているので、名称変更だけでなく移動も検知されるかと思っていましたが、私が動作確認した限りでは移動では実行されませんでした。

移動の際には「on_deleted」「on_created」「on_modified」が動作しました。つまりは、移動は「削除」→「新規作成」→「フォルダ状態が変更」と検知されているようです。この辺りの動作は、操作や環境等にもよるかもしれません。

monitor関数

monitor関数は、QuickStartの例でも見たような監視を実際に実行するために独自に作った関数です。 Observerをインスタンス化し、今回独自に作成したハンドラークラス(MyWatchHandler)を設定して、動作させています。

def monitor(path):
    """監視実行関数

    Args:
        path: 監視対象パス
    """
    event_handler = MyWatchHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

main関数

main関数は、コマンドライン引数の処理とmonitor関数の実行をしている処理です。

def main():
    """メイン関数"""
    # ===== ArgumentParserの設定
    parser = ArgumentParser(description="Monitoring Tool")
    # 引数の処理
    parser.add_argument("-p", "--path", action="store", dest="path", help="監視対象パス")
    # コマンドライン引数のパース
    args = parser.parse_args()
    # 引数の取得
    path = args.path
    # pathの指定がない場合は実行ディレクトリに設定
    if path is None:
        path = "."

    # モニター実行
    monitor(path)

コマンドライン引数の処理にはargparseを使用しています。argparseの使い方が分からない方は「argparseを用いたコマンドライン引数の取得方法」でまとめていますので興味があれば参考にしてください。

引数で対象フォルダの指定がない場合は、プログラムを実行しているフォルダが対象となります。引数を指定する場合は以下のいずれかでパスを指定します。(※プログラム名はmywatcher.pyで作成したとします。)