pyinotify

Warnings

• breaking `pyinotify` is inherently Linux-specific, as it directly interfaces with the `inotify` kernel subsystem. It will not work on other operating systems like macOS or Windows, and attempting to use it will result in errors.
  Fix: Ensure your application is deployed on a Linux environment. For cross-platform filesystem monitoring, consider libraries like `watchdog` which use different backend mechanisms.
• gotcha The `notifier.loop()` method is blocking and will prevent the rest of your application from executing. If you need non-blocking behavior (e.g., in a GUI application or when integrating with `asyncio`), you must use `notifier.process_events()` with a timeout within your own event loop or a separate thread.
  Fix: For non-blocking operations, implement a custom loop using `notifier.process_events(timeout=...)` or run the `notifier.loop()` in a dedicated thread.
• gotcha If events are generated faster than they can be processed, the kernel's inotify event queue can overflow, leading to missed events. The default queue size is usually limited (e.g., 16384 events).
  Fix: Process events promptly. Consider increasing the kernel's inotify queue size (`/proc/sys/fs/inotify/max_queued_events`) if you expect very high event volumes, but be aware of memory implications. Ensure your event handlers are efficient.
• gotcha Misunderstanding `wm.add_watch()` flags like `rec=True` (recursive) and `auto_add=True` can lead to unintended behavior, such as watching too many files/directories, or missing events for newly created directories that weren't explicitly watched.
  Fix: Carefully review the `pyinotify` documentation for `add_watch` flags. `rec=True` monitors existing subdirectories. `auto_add=True` will automatically add watches for *newly created* subdirectories, which is often desired for comprehensive recursive monitoring.

Install

• pip install pyinotify Install stable version

Imports

• WatchManager

  import pyinotify
  wm = pyinotify.WatchManager()

• Notifier

  import pyinotify
  notifier = pyinotify.Notifier(wm, event_handler)

• ProcessEvent

  import pyinotify
  class MyHandler(pyinotify.ProcessEvent): ...

Quickstart

This quickstart demonstrates how to set up a basic `pyinotify` watcher to monitor a directory for file creation and deletion events. It creates a temporary directory, sets up an `EventHandler` to print messages, and then uses `notifier.loop()` to block and process events. Remember to press Ctrl+C to stop the blocking loop. The `finally` block ensures proper cleanup of the inotify resources and the temporary directory.

import pyinotify
import os
import time

# Create a temporary directory for demonstration
watch_dir = "/tmp/pyinotify_example_dir"
os.makedirs(watch_dir, exist_ok=True)
print(f"Watching directory: {watch_dir}")
print("Try 'touch {watch_dir}/file1' or 'rm {watch_dir}/file1'")
print("Press Ctrl+C to stop.")

class EventHandler(pyinotify.ProcessEvent):
    def process_IN_CREATE(self, event):
        print(f"Created: {event.pathname}")

    def process_IN_DELETE(self, event):
        print(f"Deleted: {event.pathname}")

wm = pyinotify.WatchManager()
mask = pyinotify.IN_CREATE | pyinotify.IN_DELETE

# Attach the WatchManager and EventHandler to a Notifier
notifier = pyinotify.Notifier(wm, EventHandler())

# Add a watch to the directory for the specified events
# rec=True ensures subdirectories are also watched
wm.add_watch(watch_dir, mask, rec=True)

try:
    # Start processing events. This method blocks indefinitely.
    # For non-blocking, use notifier.process_events() in a custom loop.
    notifier.loop()
except KeyboardInterrupt:
    print("\nStopping watcher.")
finally:
    notifier.stop() # Clean up inotify resources
    # Clean up the example directory
    if os.path.exists(watch_dir):
        # Ensure directory is empty before removing
        for root, dirs, files in os.walk(watch_dir, topdown=False):
            for name in files:
                os.remove(os.path.join(root, name))
            for name in dirs:
                os.rmdir(os.path.join(root, name))
        os.rmdir(watch_dir)