Python Doc & Data Automation

Fix watchdog Event Fires Before File Is Written

The folder watcher calls the processing function from on_created, and new files fail intermittently — usually the large ones, usually when they arrive over the network:

pypdf.errors.PdfReadError: EOF marker not found
zipfile.BadZipFile: File is not a zip file              # .xlsx and .docx are ZIP archives
pandas.errors.EmptyDataError: No columns to parse from file
PermissionError: [WinError 32] The process cannot access the file because it is being used by another process

Reprocessing the same file a minute later works perfectly, which makes the bug look random.

Root Cause

Operating systems report file creation when the directory entry appears — at the moment the writer opens the new file, before any meaningful data is written. A scanner uploading a 4 MB PDF, a network copy of an Excel workbook, or an ERP streaming a CSV all trigger created within milliseconds of starting, followed by a series of modified events as data arrives. A handler that processes on created reads whatever is on disk at that instant: zero bytes, a truncated PDF without its %%EOF trailer, or half a ZIP archive whose central directory — written last — does not exist yet. On Windows the writer usually holds the file open without read sharing, so the read fails with a sharing violation instead. There is no portable event that means "the writer has finished": Linux reports closed for writers that close the file, macOS and Windows do not, and writers that rename a temporary file into place emit moved rather than created for the final name.

Minimal Diagnostic

Record what the handler sees at the moment each event fires, compared with the file's final size. The gap is the bug.

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

WATCH = Path("drop/incoming")

class Probe(FileSystemEventHandler):
    def on_any_event(self, event):
        if event.is_directory:
            return
        path = Path(getattr(event, "dest_path", "") or event.src_path)
        try:
            size_now = path.stat().st_size
            head = path.read_bytes()[:5] if size_now else b""
            tail = path.read_bytes()[-8:] if size_now else b""
        except (FileNotFoundError, PermissionError) as exc:
            size_now, head, tail = -1, b"", repr(exc).encode()[:40]
        threading.Timer(10, lambda: print(f"   {path.name} final size {path.stat().st_size}")).start()
        print(f"{event.event_type:<8} {path.name:<24} size={size_now:<9} head={head!r} tail={tail!r}")

if __name__ == "__main__":
    obs = Observer()
    obs.schedule(Probe(), str(WATCH))
    obs.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        obs.stop()
    obs.join()

created  invoice-2291.pdf         size=0         head=b'' tail=b''
modified invoice-2291.pdf         size=65536     head=b'%PDF-' tail=b'\x9e\x8a\x11...'
modified invoice-2291.pdf         size=3180544   head=b'%PDF-' tail=b'0 obj\n<<'
closed   invoice-2291.pdf         size=4213772   head=b'%PDF-' tail=b'%%EOF\n'
   invoice-2291.pdf final size 4213772

At created the file is empty; at the intermediate modified events it starts with %PDF- but ends mid-object. Only the last event shows the %%EOF trailer and the final size.

File size seen at each event When created fired the PDF was 0 bytes. At the first modified event it was 0.07 megabytes, at the second modified event 3.18 megabytes, and at the closed event 4.21 megabytes, the final size. A handler acting on created or modified reads an empty or truncated file. invoice-2291.pdf copied over SMB, bytes on disk when the event fired created 0.00 MB modified 1 0.07 MB modified 2 3.18 MB closed 4.21 MB

Fix: Enqueue on Any Event, Process Only Complete Files

Do not process inside the event handler. Queue the path on created, modified, closed and moved, and let a worker confirm the file is complete: size and modification time unchanged for a quiet period, readable without a sharing error, and — where the format allows — structurally complete. Changed lines carry comments.

# pip install "watchdog>=4.0" pypdf
import queue
import threading
import time
import zipfile
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

WATCH = Path("drop/incoming")

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

    def on_any_event(self, event):                                       # changed: never process here
        if event.is_directory or event.event_type == "deleted":
            return
        self.work.put(Path(getattr(event, "dest_path", "") or event.src_path))  # changed: moved -> dest

def structurally_complete(path: Path) -> bool:
    suffix = path.suffix.lower()
    try:
        if suffix == ".pdf":
            with path.open("rb") as fh:
                fh.seek(max(0, path.stat().st_size - 1024))
                return b"%%EOF" in fh.read()                           # changed: trailer present
        if suffix in {".xlsx", ".docx", ".pptx", ".zip"}:
            with zipfile.ZipFile(path) as zf:                           # changed: central directory readable
                return zf.testzip() is None
        return True
    except (OSError, zipfile.BadZipFile):
        return False

def is_complete(path: Path, quiet: float = 2.0, timeout: float = 900.0) -> bool:
    deadline, last, since = time.monotonic() + timeout, None, None
    while time.monotonic() < deadline:
        try:
            st = path.stat()
        except FileNotFoundError:
            return False
        current = (st.st_size, st.st_mtime_ns)
        if current != last or st.st_size == 0:
            last, since = current, None                                 # changed: still growing
        elif since is None:
            since = time.monotonic()
        elif time.monotonic() - since >= quiet:                         # changed: stable for quiet period
            try:
                with path.open("rb"):
                    pass                                                 # changed: sharing check (Windows)
            except PermissionError:
                since = None
                continue
            if structurally_complete(path):
                return True
            since = None                                                 # stable but truncated: keep waiting
        time.sleep(0.5)
    return False

def worker(work: "queue.Queue[Path]", process, stop: threading.Event) -> None:
    pending: set[Path] = set()
    while not stop.is_set():
        try:
            path = work.get(timeout=1)
        except queue.Empty:
            continue
        if path in pending:
            continue                                                     # changed: ignore event bursts
        pending.add(path)
        try:
            if is_complete(path):
                process(path)
        finally:
            pending.discard(path)

The structural check matters for writers that pause mid-transfer — a network hiccup can leave a file unchanged for longer than the quiet period while incomplete. A PDF without %%EOF near its end, or a ZIP-based Office file without a readable central directory, is treated as still in progress. CSV files have no trailer to check, so their safety comes from the quiet period alone; give slow CSV producers a longer one.

Completeness checks in order First the size must be greater than zero and unchanged together with the modification time for the quiet period. Second the file must open for reading without a PermissionError, which on Windows means the writer has released it. Third a format check must pass: PDFs need an EOF marker in the last kilobyte, and xlsx or docx files must open as ZIP archives with a readable central directory. Only then is the file processed. Stable size and mtime unchanged for the quiet period, and larger than zero bytes Opens for reading no PermissionError: on Windows the writer released its lock Format trailer PDF: %%EOF in last 1 KB; xlsx/docx: ZIP central directory readable Process once path held in a pending set so event bursts do not duplicate work

Variant Fix 1: Producers That Rename Into Place

The most reliable completeness signal is the one producers give you: write to a temporary name, then rename. A rename within one file system is atomic, so the final name only ever refers to a complete file. When you control the producer, make it do this; when you run the watcher, react to moved events for the destination name and ignore temporary names:

# stdlib only
import os
from pathlib import Path

def write_atomically(dest: Path, data: bytes) -> None:
    """Producer side: readers never see a partial dest."""
    tmp = dest.with_name(f".{dest.name}.part")
    with tmp.open("wb") as fh:
        fh.write(data)
        fh.flush()
        os.fsync(fh.fileno())                  # data on disk before the rename
    os.replace(tmp, dest)                      # atomic on the same file system

IGNORE = (".part", ".tmp", ".crdownload", ".partial", ".filepart")

def watcher_should_consider(path: Path) -> bool:
    return not path.name.startswith(".") and not path.name.endswith(IGNORE)

os.replace is atomic only within one file system. Writing the temporary file in /tmp and renaming into a watched folder on another mount becomes a copy followed by a delete — exactly the non-atomic write you were trying to avoid. Keep the temporary file in the watched directory (hidden by its leading dot) or a sibling directory on the same mount. SFTP clients such as WinSCP use .filepart suffixes by default; adding them to the ignore list turns their uploads into clean rename deliveries.

Variant Fix 2: Linux closed Events

On Linux, watchdog 3 and later report inotify's IN_CLOSE_WRITE as a closed event. It lets the stability check succeed as soon as the writer closes the file instead of after the full quiet period:

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

class FastPathOnClose(FileSystemEventHandler):
    def __init__(self, ready: "queue.Queue[Path]", maybe: "queue.Queue[Path]"):
        self.ready, self.maybe = ready, maybe

    def on_closed(self, event):                 # writer closed after writing (Linux only)
        if not event.is_directory:
            self.ready.put(Path(event.src_path))

    def on_created(self, event):
        if not event.is_directory:
            self.maybe.put(Path(event.src_path))

    def on_moved(self, event):
        if not event.is_directory:
            self.ready.put(Path(event.dest_path))

Treat closed as a strong hint, not proof: a writer that opens and closes the file several times — some scanners and sync clients do — emits several closed events, the first long before the file is complete. Still run the structural check before processing. On macOS and Windows no closed event arrives, so the quiet-period path from the main fix remains the only mechanism.

Completion signals by platform and producer On Linux, in-place writers produce a closed event, which is a strong hint, and rename writers produce a moved event, which is definitive. On macOS, in-place writers produce no completion event, so a quiet period is needed, while rename writers produce a moved event. On Windows, in-place writers produce no completion event but a sharing violation while writing, so the quiet period plus open check is needed, and rename writers produce a moved event. Format trailer checks work on all platforms. Platform In-place writer Rename writer Linux closed event (hint) moved (definitive) macOS quiet period only moved (definitive) Windows quiet + open check moved (definitive) Any format trailer check format check optional

Choosing the Quiet Period

A fixed two-second quiet period is a reasonable start but not universal. Measure how long producers actually pause between writes, and set the period per source folder:

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

def measure_write_gaps(path: Path, poll: float = 0.1, idle_stop: float = 30.0) -> dict:
    """Watch one file being written and report the longest pause between size changes."""
    sizes, gaps, last_change = [], [], time.monotonic()
    last_size = -1
    while time.monotonic() - last_change < idle_stop:
        try:
            size = path.stat().st_size
        except FileNotFoundError:
            time.sleep(poll)
            continue
        if size != last_size:
            now = time.monotonic()
            if last_size >= 0:
                gaps.append(now - last_change)
            last_change, last_size = now, size
            sizes.append(size)
        time.sleep(poll)
    return {"final_bytes": last_size, "longest_gap_s": max(gaps, default=0),
            "median_gap_s": statistics.median(gaps) if gaps else 0}

Run it while a real delivery happens from each producer — a busy scanner, an upload over a slow VPN — and set the quiet period to three or four times the longest gap observed. Record those numbers alongside the watcher configuration so the next person who sees a truncated file can compare against a measured baseline instead of guessing.

Verification

Reproduce the failure with a deliberately slow writer and assert the processing function sees the final file exactly once.

# pip install "watchdog>=4.0"
import queue
import threading
import time
from pathlib import Path
from watchdog.observers import Observer

def slow_pdf(path: Path, chunks: int = 12, pause: float = 0.4) -> int:
    body = b"%PDF-1.7\n" + b"0" * 300_000 * chunks + b"\n%%EOF\n"
    step = len(body) // chunks
    with path.open("wb") as fh:
        for i in range(0, len(body), step):
            fh.write(body[i:i + step])
            fh.flush()
            time.sleep(pause)
    return len(body)

def test_complete_only(folder: Path) -> None:
    folder.mkdir(parents=True, exist_ok=True)
    seen: list[int] = []
    work, stop = queue.Queue(), threading.Event()
    obs = Observer()
    obs.schedule(Enqueue(work), str(folder))
    obs.start()
    t = threading.Thread(target=worker, args=(work, lambda p: seen.append(p.stat().st_size), stop))
    t.start()
    expected = slow_pdf(folder / "slow.pdf")
    time.sleep(6)
    stop.set(); obs.stop(); obs.join(); t.join()
    assert seen == [expected], f"processed sizes {seen}, expected exactly [{expected}]"
    print("processed once, at full size")

if __name__ == "__main__":
    test_complete_only(Path("drop/test"))

A pause of 0.4 seconds between chunks is shorter than the two-second quiet period, so the test also confirms the worker does not fire during normal write gaps. Increase the pause beyond the quiet period to confirm the structural check catches a stalled writer.

FAQ

Can I just time.sleep(5) in on_created? It blocks the observer thread, delays every other event, and still fails for files that take longer than five seconds to copy. Queue and check instead.

Does on_modified fire once per write? No fixed rule — the OS coalesces writes differently by platform and buffer size. Treat modifications as "maybe changed", never as a count.

How do I handle files that never finish (aborted uploads)? The timeout in is_complete returns False. Move such files to a stale folder after a longer grace period so they do not block the queue or get processed later by accident.

Is checking file locks with fcntl better on Linux? Advisory locks only work if the writer takes them, which most tools do not. The quiet-period and format checks work regardless of the writer's behaviour.

Part of Watching Folders for Incoming Documents.