2 min read

Python Watchdog Module

The watchdog module provides an API and shell utilities to monitor file system events. It works cross-platform (Windows, Linux, macOS).

Installation

pip install watchdog

Importing the Module

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

Event Handlers

To react to file system events, you need to create a subclass of FileSystemEventHandler and override the methods corresponding to the events you are interested in.

Common methods:

  • on_any_event(event): Called for any event.
  • on_created(event): Called when a file or directory is created.
  • on_deleted(event): Called when a file or directory is deleted.
  • on_modified(event): Called when a file or directory is modified.
  • on_moved(event): Called when a file or directory is moved or renamed.
class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(f'File {event.src_path} has been modified')

    def on_created(self, event):
        print(f'File {event.src_path} has been created')

The Observer

The Observer class monitors the file system and dispatches events to the event handler.

if __name__ == "__main__":
    path = "."
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

Logging Example

Here is a complete example that logs all events to the console using the built-in LoggingEventHandler.

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

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()

programming/python/python