Examples

This page showcases various complete, runnable examples demonstrating different features of the watchdog API.

Pattern Matching Filter

To filter file system events using glob patterns (for example, only observing changes to .py or .pyc files), you can use the watchdog.events.PatternMatchingEventHandler:

import logging
import sys
import time

from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
from watchdog.observers import Observer

logging.basicConfig(level=logging.DEBUG)


class MyEventHandler(PatternMatchingEventHandler):
    def on_any_event(self, event: FileSystemEvent) -> None:
        logging.debug(event)


event_handler = MyEventHandler(
    patterns=["**/*.py", "**/*.pyc"], ignore_patterns=["version.py"], ignore_directories=True
)
observer = Observer()
observer.schedule(event_handler, sys.argv[1], recursive=True)
observer.start()
try:
    while True:
        time.sleep(1)
finally:
    observer.stop()
    observer.join()

Logging Trick

Watchdog also includes built-in “tricks” (pre-implemented event handlers). For instance, the watchdog.tricks.LoggerTrick automatically logs events. Here is how you can use it:

import sys
import time

from watchdog.observers import Observer
from watchdog.tricks import LoggerTrick

event_handler = LoggerTrick()
observer = Observer()
observer.schedule(event_handler, sys.argv[1], recursive=True)
observer.start()
try:
    while True:
        time.sleep(1)
finally:
    observer.stop()
    observer.join()

Debouncing Events

Text editors and build tools can emit several events for a single logical change. Use watchdog.utils.event_debouncer.EventDebouncer to collect a burst of events and run an expensive action once the watched directory has been quiet for a short interval:

from __future__ import annotations

import logging
import sys
import time

from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer
from watchdog.utils.event_debouncer import EventDebouncer

logging.basicConfig(level=logging.INFO)


def rebuild(events: list[FileSystemEvent]) -> None:
    """Run an expensive action once after a burst of file changes."""
    logging.info("Rebuilding after %d filesystem events", len(events))


class DebouncedEventHandler(FileSystemEventHandler):
    def __init__(self, debouncer: EventDebouncer) -> None:
        self.debouncer = debouncer

    def on_any_event(self, event: FileSystemEvent) -> None:
        if not event.is_directory:
            self.debouncer.handle_event(event)


path = sys.argv[1] if len(sys.argv) > 1 else "."

debouncer = EventDebouncer(debounce_interval_seconds=1, events_callback=rebuild)
event_handler = DebouncedEventHandler(debouncer)
observer = Observer()
observer.schedule(event_handler, path, recursive=True)

debouncer.start()
observer.start()
try:
    while True:
        time.sleep(1)
finally:
    observer.stop()
    debouncer.stop()
    observer.join()
    debouncer.join()

Organizing Files by Type

A common task is to keep a busy directory (such as a Downloads folder) tidy by automatically moving newly created files into category subfolders based on their file extension:

"""Organize files in a watched directory into category subfolders.

When a new file appears in the watched directory it is immediately moved
into a subfolder based on its file extension (for example ``.png`` files
are moved into ``images/`` and ``.mp4`` files into ``videos/``).

Usage::

    python file_organizer.py [path]

The directory to watch defaults to the current directory.
"""

from __future__ import annotations

import logging
import shutil
import sys
import time
from pathlib import Path

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

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%H:%M:%S")

#: Map a file extension to the subfolder it should be moved into.
CATEGORIES = {
    ".gif": "images",
    ".jpeg": "images",
    ".jpg": "images",
    ".png": "images",
    ".mkv": "videos",
    ".mov": "videos",
    ".mp4": "videos",
    ".docx": "documents",
    ".pdf": "documents",
    ".txt": "documents",
}


class FileOrganizerHandler(FileSystemEventHandler):
    def __init__(self, base_dir: Path) -> None:
        self.base_dir = base_dir

    def on_created(self, event: FileSystemEvent) -> None:
        if event.is_directory:
            return
        source = Path(event.src_path)
        category = CATEGORIES.get(source.suffix.lower(), "misc")
        target_dir = self.base_dir / category
        target_dir.mkdir(exist_ok=True)
        target = target_dir / source.name
        counter = 1
        while target.exists():
            target = target_dir / f"{source.stem}_{counter}{source.suffix}"
            counter += 1
        shutil.move(str(source), str(target))
        logging.info("Moved %s -> %s", source.name, target)


path = sys.argv[1] if len(sys.argv) > 1 else "."
base_dir = Path(path).resolve()

event_handler = FileOrganizerHandler(base_dir)
observer = Observer()
observer.schedule(event_handler, str(base_dir), recursive=False)
observer.start()
try:
    while True:
        time.sleep(1)
finally:
    observer.stop()
    observer.join()