Quickstart

Below we present a simple example that monitors the current directory recursively (which means, it will traverse any sub-directories) to detect changes. Here is what we will do with the API:

1. Create an instance of the watchdog.observers.Observer thread class.

2. Implement a subclass of watchdog.events.FileSystemEventHandler.

3. Schedule monitoring a few paths with the observer instance attaching the event handler.

4. Start the observer thread and wait for it generate events without blocking our main thread.

By default, an watchdog.observers.Observer instance will not monitor sub-directories. By passing recursive=True in the call to schedule() monitoring entire directory trees is ensured.

A Simple Example

The following example program will monitor the current directory recursively for file system changes and simply print them to the console:

from __future__ import annotations

import logging
import sys
import time

from watchdog import events
from watchdog.observers import Observer

logging.basicConfig(level=logging.DEBUG)


class MyEventHandler(events.FileSystemEventHandler):
    def on_any_event(self, event: events.FileSystemEvent) -> None:
        logging.info("Any event: %s", event)

    def on_created(self, event: events.DirCreatedEvent | events.FileCreatedEvent) -> None:
        logging.info("Created: %s", event)

    def on_deleted(self, event: events.DirDeletedEvent | events.FileDeletedEvent) -> None:
        logging.info("Deleted: %s", event)

    def on_modified(self, event: events.DirModifiedEvent | events.FileModifiedEvent) -> None:
        logging.info("Modified: %s", event)

    def on_moved(self, event: events.DirMovedEvent | events.FileMovedEvent) -> None:
        logging.info("Moved: %s", event)

    def on_opened(self, event: events.FileOpenedEvent) -> None:
        logging.info("Opened: %s", event)

    def on_closed(self, event: events.FileClosedEvent) -> None:
        logging.info("Closed: %s", event)

    def on_closed_no_write(self, event: events.FileClosedNoWriteEvent) -> None:
        logging.info("Closed no write: %s", event)


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

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

To stop the program, press Control-C.

Alternatively, you can use the observer as a context manager for cleaner code:

import time

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


class MyEventHandler(FileSystemEventHandler):
    def on_any_event(self, event: FileSystemEvent) -> None:
        print(event)


event_handler = MyEventHandler()
observer = Observer()
observer.schedule(event_handler, ".", recursive=True)

with observer:
    while True:
        time.sleep(1)

The context manager automatically handles starting and stopping the observer, ensuring proper cleanup even if an exception occurs.

Typing

If you are using type annotations it is important to note that watchdog.observers.Observer is not actually a class; it is a variable that hold the “best” observer class available on your platform.

In order to correctly type your own code your should use watchdog.observers.api.BaseObserver. For example:

from watchdog.observers import Observer
from watchdog.observers.api import BaseObserver


def my_func(obs: BaseObserver) -> None:
    # Do something with obs
    pass

observer: BaseObserver = Observer()
my_func(observer)