Watching Folders for Incoming Documents with Python

A drop folder is the simplest integration contract there is: scanners write PDFs into \\fileserver\scans, the ERP exports CSVs into /data/exports, colleagues drag Excel files into a shared "To process" directory, and the email intake saves attachments into inbox/incoming. Something has to notice each new file and start the right job. Cron jobs that run every fifteen minutes make people wait and still race with files being written. A tight while True: os.listdir() loop burns CPU and misses nothing only by accident.

Event-driven watching with the watchdog library looks like the obvious fix, and the first version works on a laptop. In production it processes half-written PDFs because the "created" event fires when the first byte lands, processes the same file three times because editors and copy tools emit several events, sees nothing at all on an SMB share, and silently skips every file that arrived while the service was restarting. This guide builds a folder watcher that avoids all four: it waits for complete files, de-duplicates events, works on network shares, catches up on startup, and hands each file exactly once to your processing code.

Prerequisites

python -m venv .venv && source .venv/bin/activate
pip install "watchdog>=4.0"
mkdir -p drop/incoming drop/processed drop/failed state

watchdog wraps the operating system's native notification APIs — inotify on Linux, FSEvents on macOS, ReadDirectoryChangesW on Windows — and offers a polling observer for file systems that do not deliver native events. The processing code itself is whatever the file needs: PDF extraction, Excel loading, Word templating. Keep it in a plain function that takes a path, so it can be tested without the watcher and reused by a batch job.

Diagnostic: Watch What the Folder Actually Emits

Before designing handling, log every raw event for a real delivery — a scanner upload, an ERP export, a file copied over the network. The event sequence differs by source, and it decides how you detect "file complete".

# pip install "watchdog>=4.0"
import logging
import sys
import time
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

WATCH = Path(sys.argv[1] if len(sys.argv) > 1 else "drop/incoming")

class LogEverything(FileSystemEventHandler):
    def on_any_event(self, event) -> None:
        size = ""
        try:
            if not event.is_directory and Path(event.src_path).exists():
                size = f"{Path(event.src_path).stat().st_size} bytes"
        except OSError:
            size = "stat failed"
        dest = f" -> {event.dest_path}" if getattr(event, "dest_path", "") else ""
        logging.info("%-9s %s%s %s", event.event_type, event.src_path, dest, size)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s.%(msecs)03d %(message)s", datefmt="%H:%M:%S")
    if not WATCH.is_dir():
        raise SystemExit(f"{WATCH} is not a directory")
    observer = Observer()
    observer.schedule(LogEverything(), str(WATCH), recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

09:14:02.118 created   drop/incoming/scan_0917.pdf 0 bytes
09:14:02.119 modified  drop/incoming/scan_0917.pdf 65536 bytes
09:14:02.524 modified  drop/incoming/scan_0917.pdf 1966080 bytes
09:14:03.201 modified  drop/incoming/scan_0917.pdf 4213772 bytes
09:14:03.202 closed    drop/incoming/scan_0917.pdf 4213772 bytes
09:14:07.880 created   drop/incoming/export.csv.tmp 0 bytes
09:14:07.991 moved     drop/incoming/export.csv.tmp -> drop/incoming/export.csv 18244 bytes

The scanner writes in place: created at zero bytes, a burst of modified, and on Linux a closed event when the writer finishes. The ERP writes a temporary name and renames it, so the final name appears only through a moved event — a watcher listening only for created never sees export.csv at all.

Core Implementation

Step 1: Turn Events into Candidate Paths

Collapse every relevant event into "this path may have changed" and put it on a queue. The handler must do almost nothing: observer threads that block on processing drop events on busy folders.

# pip install "watchdog>=4.0"
import queue
from pathlib import Path
from watchdog.events import FileSystemEventHandler

ALLOWED = {".pdf", ".csv", ".xlsx", ".docx"}
IGNORED_PREFIXES = ("~$", ".~lock", ".")                    # Office and LibreOffice lock files, hidden files
IGNORED_SUFFIXES = (".tmp", ".part", ".crdownload", ".partial")

def interesting(path: Path) -> bool:
    return (path.suffix.lower() in ALLOWED
            and not path.name.startswith(IGNORED_PREFIXES)
            and not path.name.endswith(IGNORED_SUFFIXES))

class Enqueue(FileSystemEventHandler):
    def __init__(self, work: "queue.Queue[Path]") -> None:
        self.work = work

    def _offer(self, raw: str) -> None:
        path = Path(raw)
        if interesting(path):
            self.work.put(path)

    def on_created(self, event):
        if not event.is_directory:
            self._offer(event.src_path)

    def on_modified(self, event):
        if not event.is_directory:
            self._offer(event.src_path)

    def on_closed(self, event):                             # Linux inotify: writer closed the file
        if not event.is_directory:
            self._offer(event.src_path)

    def on_moved(self, event):                              # temp-then-rename writers
        if not event.is_directory:
            self._offer(event.dest_path)

Ignoring temporary suffixes and lock-file prefixes at the handler level keeps noise off the queue. Handling on_moved with the destination path is what makes rename-based writers work.

Step 2: Wait Until the File Is Complete

No single event means "complete" on every platform. A robust rule combines two checks: the size and modification time have been stable for a quiet period, and the file can be opened for reading without a sharing error (Windows) or is not open for writing (where you can check).

# stdlib only
import os
import time
from pathlib import Path

def wait_until_stable(path: Path, quiet: float = 2.0, timeout: float = 600.0) -> bool:
    """True once size and mtime have not changed for `quiet` seconds; False if gone or timed out."""
    deadline = time.monotonic() + timeout
    last, stable_since = None, None
    while time.monotonic() < deadline:
        try:
            st = path.stat()
        except FileNotFoundError:
            return False                                     # moved away or deleted
        current = (st.st_size, st.st_mtime_ns)
        if current == last and st.st_size > 0:
            if stable_since is None:
                stable_since = time.monotonic()
            if time.monotonic() - stable_since >= quiet:
                try:
                    with path.open("rb") as fh:              # Windows raises if a writer holds a lock
                        fh.read(1)
                    return True
                except PermissionError:
                    stable_since = None
        else:
            last, stable_since = current, None
        time.sleep(0.5)
    return False

The quiet period is a trade-off between latency and safety: two seconds suits local scanners and exports; network copies of large files can pause longer between writes and need more. The event-specific details, including format checks such as a PDF's %%EOF trailer, are in fix watchdog event fires before file is written.

Step 3: One Worker, De-duplicated, Exactly Once

Events for the same file arrive many times. A single worker thread drains the queue, skips paths already being handled or done, waits for stability, and records completion in a small state store keyed on the file's content hash.

# stdlib only
import hashlib
import logging
import queue
import sqlite3
import threading
from pathlib import Path

log = logging.getLogger("watcher")

def file_hash(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as fh:
        for block in iter(lambda: fh.read(1 << 20), b""):
            h.update(block)
    return h.hexdigest()

def worker(work: "queue.Queue[Path]", process, stop: threading.Event, db_path: Path) -> None:
    db = sqlite3.connect(db_path)
    db.execute("CREATE TABLE IF NOT EXISTS done (sha256 TEXT PRIMARY KEY, name TEXT, at TEXT DEFAULT CURRENT_TIMESTAMP)")
    in_flight: set[Path] = set()
    while not stop.is_set():
        try:
            path = work.get(timeout=1)
        except queue.Empty:
            continue
        if path in in_flight or not path.exists():
            continue
        in_flight.add(path)
        try:
            if not wait_until_stable(path):
                log.warning("%s never became stable or disappeared", path.name)
                continue
            digest = file_hash(path)
            if db.execute("SELECT 1 FROM done WHERE sha256 = ?", (digest,)).fetchone():
                log.info("%s already processed (same content)", path.name)
                continue
            process(path)                                   # your document job
            db.execute("INSERT INTO done (sha256, name) VALUES (?, ?)", (digest, path.name))
            db.commit()
        except Exception:
            log.exception("processing failed for %s", path.name)
        finally:
            in_flight.discard(path)
    db.close()

A single worker is deliberate. Most document jobs are fast relative to arrival rates, and one worker makes ordering and de-duplication trivial. If processing is slow, keep one dispatcher doing the de-duplication and stability check, and hand stable paths to a process pool.

Step 4: Catch Up on Startup and Tolerate Missed Events

Files that arrive while the watcher is stopped produce no events when it starts again, and native event queues can overflow during large bursts. Scan the folder on startup and periodically, feeding the same queue — de-duplication makes it harmless to offer a path twice.

# pip install "watchdog>=4.0"
import threading
import time
from pathlib import Path

def rescan(folder: Path, work, stop: threading.Event, every: float = 300.0) -> None:
    while not stop.is_set():
        for path in sorted(folder.iterdir(), key=lambda p: p.stat().st_mtime if p.exists() else 0):
            if path.is_file() and interesting(path):
                work.put(path)
        stop.wait(every)

Processing oldest first during catch-up keeps downstream ordering sensible — yesterday's statement before today's. The periodic rescan is also the safety net for anything the native API silently dropped.

Step 5: Move Files Out When Done

A drop folder should only contain work that has not been done. After processing, move each file to processed/ or, on failure, to failed/ with a reason, using date-based subfolders and collision-safe names. The patterns are in move processed files to archive folders.

Edge Cases and Variants

Network Shares and Mounted Cloud Drives

SMB and NFS mounts rarely deliver change notifications to the client machine, and cloud-sync folders deliver them inconsistently. Use watchdog's polling observer for those paths — it compares directory snapshots on an interval and emits the same events:

# pip install "watchdog>=4.0"
from watchdog.observers.polling import PollingObserver

def network_observer(path: str, handler, interval: float = 5.0) -> PollingObserver:
    observer = PollingObserver(timeout=interval)          # snapshot comparison every `interval` seconds
    observer.schedule(handler, path, recursive=False)
    return observer

Polling large directories is expensive over the network; keep drop folders small by moving processed files out promptly, and prefer running the watcher on the file server itself where native events work.

Many Subfolders

recursive=True watches a tree, but each subdirectory consumes an inotify watch on Linux, and the default limit (fs.inotify.max_user_watches) can be exhausted on large trees, producing OSError: [Errno 28] inotify watch limit reached. Raise the limit or restructure so producers write to one flat drop folder.

Running as a Service

A watcher is a long-running process and needs to start at boot, restart on failure and log somewhere durable. Run a document watcher as a systemd service covers the unit file, graceful shutdown and journald logging.

Validation

Test the watcher against the delivery patterns you saw in the diagnostic, with deliberately slow writes, before trusting it with real documents.

# pip install "watchdog>=4.0"
import threading
import time
from pathlib import Path

def slow_write(path: Path, total: int = 3_000_000, chunk: int = 250_000, pause: float = 0.3) -> None:
    with path.open("wb") as fh:
        fh.write(b"%PDF-1.7\n")
        for _ in range(total // chunk):
            fh.write(b"0" * chunk)
            fh.flush()
            time.sleep(pause)
        fh.write(b"\n%%EOF\n")

def test_watcher(folder: Path) -> None:
    seen: list[tuple[str, int]] = []
    def process(path: Path) -> None:
        seen.append((path.name, path.stat().st_size))
    work, stop = __import__("queue").Queue(), threading.Event()
    # start observer + worker exactly as the service does (omitted: same code as Steps 1-4)
    start_watcher(folder, work, process, stop)
    slow_write(folder / "slow.pdf")
    (folder / "renamed.csv.tmp").write_text("a,b\n1,2\n")
    (folder / "renamed.csv.tmp").rename(folder / "renamed.csv")
    time.sleep(8)
    stop.set()
    names = [n for n, _ in seen]
    assert names.count("slow.pdf") == 1, f"slow.pdf processed {names.count('slow.pdf')} times"
    assert dict(seen)["slow.pdf"] > 3_000_000, "slow.pdf processed before it was complete"
    assert "renamed.csv" in names, "rename-based delivery missed"
    print("watcher handles slow writes and renames exactly once")

The three assertions map directly to the production failures: processed more than once, processed while incomplete, and missed entirely. Run the test on the same kind of storage the watcher will use — passing on a local SSD proves nothing about an NFS mount.

Performance and Scale Notes

Native observers cost almost nothing while idle and deliver events within milliseconds. The worker's stability wait dominates latency by design; tune the quiet period per source rather than globally if some producers are slow. Hashing large files for de-duplication costs about a second per gigabyte on typical disks — acceptable for documents, but key on name, size and modification time instead for multi-gigabyte media. Polling observers scale with the number of entries in the watched directory, so a folder that accumulates 50,000 processed files makes every poll slow; moving files out is a performance requirement, not just tidiness. A single watcher comfortably handles thousands of files per hour when processing is fast; beyond that, separate the watcher (enqueue only) from processing workers connected by a proper queue.

Troubleshooting

Complete Working Script

#!/usr/bin/env python3
# pip install "watchdog>=4.0"
"""Watch a drop folder and process each complete document exactly once."""
import argparse
import hashlib
import logging
import queue
import shutil
import signal
import sqlite3
import threading
import time
from datetime import date
from pathlib import Path

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
from watchdog.observers.polling import PollingObserver

ALLOWED = {".pdf", ".csv", ".xlsx", ".docx"}
log = logging.getLogger("watcher")


def interesting(p: Path) -> bool:
    return p.suffix.lower() in ALLOWED and not p.name.startswith(("~$", ".")) and not p.name.endswith((".tmp", ".part"))


class Enqueue(FileSystemEventHandler):
    def __init__(self, work):
        self.work = work

    def on_any_event(self, event):
        if event.is_directory:
            return
        raw = getattr(event, "dest_path", "") or event.src_path
        if event.event_type in ("created", "modified", "closed", "moved") and interesting(Path(raw)):
            self.work.put(Path(raw))


def stable(p: Path, quiet: float) -> bool:
    last, since = None, None
    for _ in range(int(600 / 0.5)):
        try:
            st = p.stat()
        except FileNotFoundError:
            return False
        cur = (st.st_size, st.st_mtime_ns)
        if cur == last and st.st_size > 0:
            since = since or time.monotonic()
            if time.monotonic() - since >= quiet:
                return True
        else:
            last, since = cur, None
        time.sleep(0.5)
    return False


def process(path: Path) -> None:
    log.info("processing %s (%d bytes)", path.name, path.stat().st_size)   # replace with the real job


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("folder", type=Path)
    ap.add_argument("--poll", action="store_true", help="use polling (network shares)")
    ap.add_argument("--quiet", type=float, default=2.0)
    args = ap.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    root = args.folder.resolve()
    done_dir, failed_dir, state = root.parent / "processed", root.parent / "failed", root.parent / "state"
    for d in (done_dir, failed_dir, state):
        d.mkdir(parents=True, exist_ok=True)

    work: "queue.Queue[Path]" = queue.Queue()
    stop = threading.Event()
    signal.signal(signal.SIGTERM, lambda *_: stop.set())
    observer = (PollingObserver(timeout=5) if args.poll else Observer())
    observer.schedule(Enqueue(work), str(root), recursive=False)
    observer.start()
    for p in sorted(root.iterdir(), key=lambda x: x.stat().st_mtime):
        if p.is_file() and interesting(p):
            work.put(p)                                             # startup catch-up

    db = sqlite3.connect(state / "watcher.sqlite")
    db.execute("CREATE TABLE IF NOT EXISTS done (sha256 TEXT PRIMARY KEY, name TEXT)")
    try:
        while not stop.is_set():
            try:
                path = work.get(timeout=1)
            except queue.Empty:
                continue
            if not path.exists() or not stable(path, args.quiet):
                continue
            digest = hashlib.sha256(path.read_bytes()).hexdigest()
            target = done_dir / date.today().isoformat()
            if not db.execute("SELECT 1 FROM done WHERE sha256=?", (digest,)).fetchone():
                try:
                    process(path)
                    db.execute("INSERT INTO done VALUES (?, ?)", (digest, path.name))
                    db.commit()
                except Exception:
                    log.exception("failed: %s", path.name)
                    target = failed_dir / date.today().isoformat()
            target.mkdir(parents=True, exist_ok=True)
            shutil.move(str(path), str(target / f"{digest[:8]}_{path.name}"))
    except KeyboardInterrupt:
        pass
    finally:
        observer.stop()
        observer.join()
        db.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Frequently Asked Questions

Should I use watchdog or a scheduled batch job? Use a scheduled job when a delay of minutes is acceptable and the folder is on a network share — it is simpler and restarts cleanly. Use a watcher when people wait for results, such as a scan-to-searchable-PDF service.

Is on_closed reliable enough to skip the stability check? Only on Linux with inotify, and only for writers that close once. Keep the stability check; closed simply lets it succeed sooner.

Can several watchers share one folder? Not safely without locking — two processes will both pick up a file. Run one watcher per folder, or claim files by atomically renaming them into a per-worker directory first.

How do I watch an S3 bucket or SharePoint library? Those are not file systems. Use the service's event notifications (S3 events, Microsoft Graph change notifications) or poll their APIs; the de-duplication and stability ideas still apply.

Related

• Fix watchdog Event Fires Before File Is Written — completeness checks per platform and format
• Move Processed Files to Archive Folders — safe moves, collisions and retention
• Run a Document Watcher as a systemd Service — keeping the watcher running
• Processing Email Attachments Automatically — a common producer for the drop folder
• Scheduling and Logging Automation Jobs — the batch alternative to watching

Part of Automating Document & Data Pipelines.