Building a Watch Mode with watchfiles in Python

Your CLI builds something from source files — a documentation site, a set of rendered templates, generated code, a bundle of config — and developers run it dozens of times an hour while editing. A --watch flag that rebuilds automatically on save is one of the highest-value features a developer tool can offer. Written naively, it is also one of the most irritating: rebuilding five times per save, rebuilding forever because the build's own output triggers the watcher, exiting on the first syntax error, or burning a CPU core polling the disk. This guide builds a --watch mode on the watchfiles package that batches changes, ignores what it should, survives broken input, prints useful output and can be tested deterministically. It belongs to the long-running and watch-mode topic.

Prerequisites

• Python 3.10+, Typer and watchfiles 1.x (uv add watchfiles). It ships prebuilt wheels for Linux, macOS and Windows.
• A build function you can call repeatedly: build(src: Path, out: Path) -> Result.
• Familiarity with the loop and stop-event ideas on the topic overview.

How change detection works

Modern operating systems notify programs when files change — inotify on Linux, FSEvents on macOS, ReadDirectoryChangesW on Windows — so a watcher does not need to poll. The catch is that one logical "save" is rarely one event. Editors write to a temporary file and rename it over the original, touch backup and swap files, and update metadata separately. A single Ctrl+S in some editors produces five or more events within a few milliseconds.

watchfiles wraps the native APIs through Rust's notify library and debounces: it waits until events stop arriving for a short step (50 ms by default), up to a maximum window, then yields everything collected as one set. Your loop sees one batch per save, not five events.

The recipe

# src/mytool/watching.py
from __future__ import annotations

import threading
import time
from collections.abc import Callable
from datetime import datetime
from pathlib import Path

from watchfiles import Change, DefaultFilter, watch

Echo = Callable[[str], None]


def _stamp() -> str:
    return datetime.now().strftime("%H:%M:%S")


def _describe(changes: set[tuple[Change, str]], root: Path) -> str:
    paths = sorted({Path(p) for _, p in changes})
    shown = ", ".join(str(p.relative_to(root)) if p.is_relative_to(root) else str(p) for p in paths[:3])
    more = f" (+{len(paths) - 3} more)" if len(paths) > 3 else ""
    return shown + more


def watch_and_rebuild(src: Path, out: Path, build: Callable[[set[Path] | None], int], *,
                      echo: Echo, stop: threading.Event | None = None,
                      extensions: tuple[str, ...] = (".md", ".html", ".toml")) -> None:
    src, out = src.resolve(), out.resolve()

    class SourceOnly(DefaultFilter):
        def __call__(self, change: Change, path: str) -> bool:
            return super().__call__(change, path) and path.endswith(extensions)

    watch_filter = SourceOnly(ignore_paths=[out])     # our own output never triggers a build

    def run(changed: set[Path] | None) -> None:
        start = time.perf_counter()
        try:
            count = build(changed)
        except Exception as exc:                        # a broken file must not end the watch
            echo(f"[{_stamp()}] error: {exc} — still watching")
            return
        echo(f"[{_stamp()}] built {count} file(s) in {time.perf_counter() - start:.2f}s")

    run(None)                                           # full build first
    echo(f"[{_stamp()}] watching {src} (Ctrl+C to stop)")
    for changes in watch(src, watch_filter=watch_filter, stop_event=stop, debounce=400):
        echo(f"[{_stamp()}] changed: {_describe(changes, src)}")
        deleted = {Path(p) for c, p in changes if c == Change.deleted}
        run(None if deleted else {Path(p) for _, p in changes})

# src/mytool/cli.py
from pathlib import Path

import typer

from mytool.watching import watch_and_rebuild

app = typer.Typer()


def build_site(src: Path, out: Path, changed: set[Path] | None) -> int:
    """Render Markdown to HTML; only the changed files when we know them."""
    targets = sorted(changed) if changed else sorted(src.rglob("*.md"))
    for md in targets:
        if md.suffix != ".md":
            continue
        text = md.read_text(encoding="utf-8")
        if "{{" in text and "}}" not in text:
            raise ValueError(f"{md.name}: unclosed template tag")
        dest = out / md.relative_to(src).with_suffix(".html")
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_text(f"<article>{text}</article>\n", encoding="utf-8")
    return len(targets)


@app.callback()
def main() -> None:
    """Static site tools."""


@app.command()
def build(
    src: Path = typer.Argument(Path("docs"), exists=True, file_okay=False),
    out: Path = typer.Option(Path("site"), "--out", "-o"),
    watch: bool = typer.Option(False, "--watch", "-w", help="Rebuild when sources change."),
) -> None:
    """Build the site once, or keep rebuilding with --watch."""
    src, out = src.resolve(), out.resolve()
    if not watch:
        n = build_site(src, out, None)
        typer.echo(f"built {n} file(s)", err=True)
        return
    try:
        watch_and_rebuild(src, out, lambda changed: build_site(src, out, changed),
                          echo=lambda msg: typer.echo(msg, err=True))
    except KeyboardInterrupt:
        typer.echo("stopped watching", err=True)


if __name__ == "__main__":
    app()

The decisions that make it pleasant

Ignore your own output. If site/ sits inside docs/, or the build writes caches next to sources, every build produces events that trigger the next build — an infinite loop that pins the CPU. ignore_paths=[out] removes that failure mode entirely. DefaultFilter already ignores .git, .venv, node_modules, __pycache__, .pyc files and editor swap files.

Filter by extension. Restricting to the file types the build actually reads keeps unrelated edits — a README, a notes file — from triggering rebuilds.

Build everything first. The initial full build guarantees the output is current before watching begins, so the first save does not reveal a stale site.

Rebuild incrementally when it is safe. Passing the set of changed paths lets the build skip untouched files, which is what keeps rebuilds under a second on large projects. Deletions and renames are harder to handle incrementally — the old output must be removed — so a batch containing a deletion falls back to a full rebuild. When in doubt, full rebuilds are correct; incremental ones are an optimisation.

Errors are output, not exits. A developer mid-edit will save broken files constantly. The watch loop catches exceptions from the build, prints them with a timestamp, and waits for the next save. Exiting on the first error would make watch mode useless.

Stop events for tests and embedding. watch() accepts a stop_event, which ends the iteration cleanly when set. The CLI does not need it — Ctrl+C raises KeyboardInterrupt out of the generator — but tests and any code embedding the watcher do.

UX considerations

• Timestamp every line. When a developer glances at the terminal, the time tells them whether the last build reflects their latest save.
• Name what changed. "changed: docs/intro.md" confirms the watcher saw the right file — the most common watch-mode confusion is editing a file outside the watched directory.
• Keep it quiet. One line for the change and one for the result is enough. Stream full build logs only with --verbose.
• Clear the screen optionally. Some tools clear the terminal before each rebuild so only current errors are visible. Offer it as --clear; do not force it, because it destroys scrollback people may want.
• Combine with a server carefully. If the command also serves the output (--serve), run the HTTP server in a thread and the watch loop in the main thread, so Ctrl+C reaches the watcher.
• Mention polling for odd filesystems. Network drives, Docker bind mounts on macOS and some WSL paths do not deliver native events. watchfiles supports force_polling=True (or the WATCHFILES_FORCE_POLLING environment variable); document it for users who see no rebuilds.

Testing the behaviour

Watch mode is testable without sleeps-and-hope: run the watcher in a background thread with a stop_event, make a real file change, wait for the build callback, then stop:

# tests/test_watch.py
import threading
from pathlib import Path

from mytool.watching import watch_and_rebuild


def run_watcher(src: Path, out: Path):
    builds: list[set[Path] | None] = []
    built = threading.Event()
    stop = threading.Event()
    messages: list[str] = []

    def build(changed):
        builds.append(changed)
        built.set()
        if changed and any(p.name == "broken.md" for p in changed):
            raise ValueError("unclosed tag")
        return len(changed or [])

    t = threading.Thread(target=watch_and_rebuild, args=(src, out, build),
                         kwargs={"echo": messages.append, "stop": stop}, daemon=True)
    t.start()
    return builds, built, stop, messages, t


def test_initial_build_then_rebuild_on_change(tmp_path):
    src, out = tmp_path / "docs", tmp_path / "site"
    src.mkdir()
    (src / "a.md").write_text("hello")
    builds, built, stop, messages, t = run_watcher(src, out)
    assert built.wait(5) and builds[0] is None          # full initial build
    built.clear()
    (src / "a.md").write_text("hello again")
    assert built.wait(5)
    assert builds[-1] == {(src / "a.md").resolve()}
    stop.set()
    t.join(5)


def test_output_dir_is_ignored_and_errors_do_not_stop(tmp_path):
    src = tmp_path / "docs"
    out = src / "_site"                                   # output inside the source tree
    out.mkdir(parents=True)
    builds, built, stop, messages, t = run_watcher(src, out)
    assert built.wait(5)
    built.clear()
    (out / "index.html").write_text("generated")          # must NOT trigger a build
    assert not built.wait(1.5)
    (src / "broken.md").write_text("{{ oops")
    assert built.wait(5)
    assert any("still watching" in m for m in messages)
    assert t.is_alive()                                   # the error did not end the watch
    stop.set()
    t.join(5)

These tests touch the real filesystem and real OS notifications, so they are a little slower than pure unit tests — a few seconds in total — but they catch exactly the regressions that matter: a lost ignore rule and an exception escaping the loop. If a CI environment lacks native events, set WATCHFILES_FORCE_POLLING=1 for the test job.

Conclusion

A good watch mode is a loop with four properties: one rebuild per save (debouncing), no rebuilds from its own output (ignore paths), no exit on broken input (catch and report), and output that tells the developer what changed and when. watchfiles provides the first two almost for free, and the rest is a dozen lines of careful loop. Test it with a stop event and real file writes, and it becomes a feature your users reach for every day.

Frequently asked questions

Related

• Up: Long-running and watch-mode CLIs
• Handling SIGTERM and graceful shutdown
• Health checks and heartbeats for long-running CLIs
• Caching expensive work between CLI runs
• Detecting a TTY and adapting output