watchdog

---

1. Monitor a Thread

import threading
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

def watch_thread(thread):
    class MyEventHandler(PatternMatchingEventHandler):
        patterns = ["*"]

        def on_modified(self, event):
            # Restart the thread if it's not running
            if not thread.is_alive():
                thread.start()

        def on_created(self, event):
            # Restart the thread if it's not running
            if not thread.is_alive():
                thread.start()

    observer = Observer()
    observer.schedule(MyEventHandler(), thread.name, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

thread = threading.Thread(target=my_function)
watch_thread(thread)

2. Monitor a File

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

def watch_file(file_path):
    class MyEventHandler(PatternMatchingEventHandler):
        patterns = [file_path]

        def on_modified(self, event):
            # Perform some action when the file is modified
            print("File modified: {}".format(event.src_path))

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

watch_file("my_file.txt")

3. Monitor a Directory

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

def watch_directory(dir_path):
    class MyEventHandler(FileSystemEventHandler):
        def on_modified(self, event):
            # Perform some action when a file in the directory is modified
            print("File modified: {}".format(event.src_path))

        def on_created(self, event):
            # Perform some action when a new file is created in the directory
            print("File created: {}".format(event.src_path))

        def on_deleted(self, event):
            # Perform some action when a file is deleted from the directory
            print("File deleted: {}".format(event.src_path))

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

watch_directory("my_directory")

4. Monitor a Regex Pattern

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

def watch_regex(dir_path, pattern):
    class MyEventHandler(PatternMatchingEventHandler):
        patterns = [pattern]

        def on_modified(self, event):
            # Perform some action when a file matching the regex is modified
            print("File modified: {}".format(event.src_path))

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

watch_regex("my_directory", ".*\.py$")

5. Monitor Multiple Directories

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

def watch_multiple_directories(dir_paths):
    class MyEventHandler(PatternMatchingEventHandler):
        patterns = ["*"]

        def on_modified(self, event):
            # Perform some action when a file in any of the directories is modified
            print("File modified: {}".format(event.src_path))

    observer = Observer()
    for dir_path in dir_paths:
        observer.schedule(MyEventHandler(), dir_path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

watch_multiple_directories(["my_directory1", "my_directory2"])

6. Ignore Hidden Files and Directories

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

def watch_ignore_hidden(dir_path):
    class MyEventHandler(PatternMatchingEventHandler):
        patterns = ["*"]
        ignore_directories = True
        ignore_hidden_files = True

        def on_modified(self, event):
            # Perform some action when a non-hidden file is modified
            print("File modified: {}".format(event.src_path))

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

watch_ignore_hidden("my_directory")

7. Monitor File Events and Errors

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

def watch_events_errors(dir_path):
    class MyEventHandler(PatternMatchingEventHandler, FileSystemEventHandler):
        patterns = ["*"]

        def on_modified(self, event):
            # Perform some action when a file is modified
            print("File modified: {}".format(event.src_path))

        def on_error(self, event):
            # Handle any errors that occur during event processing
            print("Error: {}".format(event))

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

watch_events_errors("my_directory")

8. Monitor a File Handle

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

def watch_file_handle(file_handle):
    class MyEventHandler(FileModifiedHandler):
        def on_modified(self, event):
            # Perform some action when the file handle is modified
            print("File modified: {}".format(event.src_path))

    observer = Observer()
    observer.schedule(MyEventHandler(), file_handle, recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

with open("my_file.txt", "w") as file_handle:
    watch_file_handle(file_handle)

9. Monitor File Attributes

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

def watch_attributes(dir_path):
    class MyEventHandler(FileSystemEventHandler):
        def on_modified(self, event):
            # Perform some action when the file's attributes are modified
            print("File modified: {}".format(event.src_path))
            print("File attributes: {}".format(event.stat))

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

watch_attributes("my_directory")

10. Monitor File Move Events

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

def watch_move_events(dir_path):
    class MyEventHandler(FileSystemEventHandler):
        def on_moved(self, event):
            # Perform some action when a file is moved
            print("File moved: {} -> {}".format(event.src_path, event.dest_path))

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

watch_move_events("my_directory")