Watch a Folder and Auto-Process New Files with Python (watchdog)

The watchdog library monitors cross-platform file system events using native OS APIs (inotify on Linux, FSEvents on macOS, ReadDirectoryChangesW on Windows). To automatically process files the moment they land in a target directory, implement a custom FileSystemEventHandler and attach it to an Observer thread.

1. Installation

Install the library using pip:

pip install watchdog

2. Complete Python Monitoring Script

The following script watches a designated directory, filters out subdirectories, waits for file write operations to complete, and triggers a custom handler function.

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

class NewFileHandler(FileSystemEventHandler):
    def on_created(self, event):
        # Ignore subdirectories
        if event.is_directory:
            return

        file_path = event.src_path
        print(f"New file detected: {file_path}")
        self.process_file(file_path)

    def process_file(self, file_path):
        """Custom processing logic goes here."""
        try:
            # Wait until the writing process finishes releasing the file
            if self._wait_for_file_ready(file_path):
                print(f"Processing: {file_path}")
                # Add parsing, database inserts, or file movement logic here
                print(f"Finished processing: {file_path}")
            else:
                print(f"Timeout: File {file_path} was not ready in time.")
        except Exception as e:
            print(f"Error processing {file_path}: {e}")

    def _wait_for_file_ready(self, file_path, timeout=10):
        """Ensures the file is fully written before processing."""
        start_time = time.time()
        last_size = -1

        while time.time() - start_time < timeout:
            if os.path.exists(file_path):
                current_size = os.path.getsize(file_path)
                # Check if file size has stabilized and is not 0
                if current_size == last_size and current_size > 0:
                    return True
                last_size = current_size
            time.sleep(0.5)

        return False

def main():
    watch_directory = "./watch_folder"

    if not os.path.exists(watch_directory):
        os.makedirs(watch_directory)

    event_handler = NewFileHandler()
    observer = Observer()
    observer.schedule(event_handler, path=watch_directory, recursive=False)

    print(f"Monitoring started for directory: {watch_directory}")
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("Monitoring stopped.")

    observer.join()

if __name__ == "__main__":
    main()

3. Key Components Explained

• FileSystemEventHandler: Handles incoming events. Overriding on_created isolates triggers specifically for newly added items.
• Observer Thread: Runs in the background and delegates file system actions to your handler without blocking the main program execution loop.
• event.is_directory Check: Prevents accidental execution of file handling logic on newly created subfolders.
• recursive=False: Restricts watching strictly to the top-level directory. Change to True to include nested directories.

4. Handling Common Edge Cases

• Partial Writes / File Locking: The on_created event fires instantly when a file is created, often before a copy or download operation finishes. The _wait_for_file_ready method avoids reading incomplete data by polling the file size until it stabilizes.
• Infinite Loops: If your script writes output files back into the monitored directory, it will re-trigger the handler. Always save processed outputs to an outside folder or filter allowed file extensions explicitly inside on_created.
• Atomic Moves: Applications like file transfer clients often drop temporary files (e.g., .tmp) and rename them upon completion. To catch these, extend your class to override on_moved alongside on_created.

Need this done fast? order this automation on FreelanceHunt.