watchdog-gevent

import time import os import gevent.monkey from watchdog.events import FileSystemEventHandler from watchdog_gevent import Observer # IMPORTANT: Apply gevent monkey patching BEFORE importing other blocking libraries gevent.monkey.patch_all() class MyHandler(FileSystemEventHandler): def on_created(self, event): print(f"Gevent: File created: {event.src_path}") def on_deleted(self, event): print(f"Gevent: File deleted: {event.src_path}") def on_modified(self, event): print(f"Gevent: File modified: {event.src_path}") # Note: watchdog-gevent does not emit on_moved events. # def on_moved(self, event): # print(f"Gevent: File moved from {event.src_path} to {event.dest_path}") if __name__ == "__main__": # Create a temporary directory for demonstration path = os.path.join(os.getcwd(), "gevent_test_dir") os.makedirs(path, exist_ok=True) print(f"Monitoring directory: {path}") event_handler = MyHandler() observer = Observer() observer.schedule(event_handler, path, recursive=True) observer.start() try: while True: time.sleep(1) # gevent.sleep(1) would also work after monkey-patching except KeyboardInterrupt: observer.stop() observer.join() print("Observer stopped.") # Clean up os.rmdir(path)