Intermediate
You are building a development tool, a config reloader, or a file-processing pipeline. You want it to react instantly when a file changes — no manual restarts, no polling in a tight loop burning CPU. The classic approach was watchdog, but uvicorn, FastAPI’s dev server, and several modern Python tools have moved to watchfiles instead. It is faster, simpler, and backed by a Rust native-event library that uses zero-overhead OS file system notifications.
Python’s watchfiles library wraps the Rust notify crate to deliver real-time change events on Linux (inotify), macOS (FSEvents), and Windows (ReadDirectoryChangesW). It works with both synchronous and async Python, requires a single pip install, and has no system-level dependencies to manage. If you have used watchdog, you will find watchfiles achieves the same result with a much smaller API surface.
This article covers the core watch() and awatch() functions, filtering by file type, working with async code using awatch(), integrating with FastAPI for a live-reload pattern, and a real-life config-watcher that restarts a service when its YAML file changes. By the end you will know how to trigger any Python code the moment a file on disk is modified.
Python watchfiles: Quick Example
Here is the minimal working example — a blocking watcher that prints every change to the current directory. Run this in one terminal, then edit or create a file in another terminal to see events arrive.
# quick_watchfiles.py
from watchfiles import watch
print("Watching current directory. Press Ctrl+C to stop.")
for changes in watch("."):
for change_type, path in changes:
print(f"{change_type.name}: {path}")
Output (after editing a file called config.yml):
Watching current directory. Press Ctrl+C to stop.
modified: /home/user/project/config.yml
modified: /home/user/project/config.yml
added: /home/user/project/notes.txt
The watch() function blocks and yields a set of (Change, path) tuples each time any file in the watched path changes. Change is an enum with three values: Change.added, Change.modified, and Change.deleted. The loop body runs once per batch of changes — if 10 files change simultaneously, you get one iteration with 10 tuples, not 10 separate iterations.
The sections below cover filtering, async usage, debouncing, and how to build a production-grade auto-reloader with this foundation.
What Is watchfiles and Why Use It?
watchfiles is a Python wrapper around the Rust notify library, written by Samuel Colvin (the creator of Pydantic). It uses native OS file-system event APIs — inotify on Linux, FSEvents on macOS, and ReadDirectoryChangesW on Windows — instead of polling. This means it reacts in milliseconds without burning CPU in a busy loop.
The key difference from polling-based alternatives: polling wakes up every N milliseconds and checks whether any file modification time has changed. Native events arrive from the OS exactly when a change happens. For high-frequency workflows like hot-reload on every keystroke, the difference in CPU load is significant.
| Feature | polling (manual) | watchdog | watchfiles |
|---|---|---|---|
| Event source | os.stat() loop | OS native + fallback | Rust notify (OS native) |
| CPU idle load | High | Low | Very low |
| async support | Manual | Via threads | Built-in (awatch) |
| API complexity | High | Medium (handlers) | Low (iterator) |
| Debouncing | Manual | Manual | Built-in (100ms default) |
| pip install | N/A | watchdog | watchfiles |
Install before running any of the examples below:
# In your terminal
pip install watchfiles
The package ships pre-compiled Rust binaries for all common platforms, so there is no Rust toolchain needed on your machine. The install is a single .whl download.
The watch() Function
The synchronous watch() function is an iterator that blocks until changes occur, then yields a set of change events. It accepts one or more paths and an optional set of keyword arguments for filtering and debouncing.
# watch_basics.py
from watchfiles import watch, Change
# Watch multiple directories simultaneously
for changes in watch("./src", "./config"):
for change_type, path in changes:
if change_type == Change.added:
print(f"[+] New file: {path}")
elif change_type == Change.modified:
print(f"[~] Changed: {path}")
elif change_type == Change.deleted:
print(f"[-] Deleted: {path}")
Output (after creating and modifying a file):
[+] New file: /project/config/new_rule.yml
[~] Changed: /project/src/main.py
[-] Deleted: /project/src/old_module.py
Passing multiple paths as separate arguments watches all of them with a single thread — no need to spin up multiple watchers. The yielded set may contain changes from any of the watched paths in a single batch.
Stopping the Watcher Programmatically
By default, watch() runs until interrupted. To stop it after a condition is met, call stop_event — a threading event you pass in and set from another thread when you want the watcher to exit.
# watch_stop.py
import threading
from watchfiles import watch
stop = threading.Event()
def watcher():
for changes in watch(".", stop_event=stop):
for change_type, path in changes:
print(f"{change_type.name}: {path}")
print("Watcher stopped.")
t = threading.Thread(target=watcher, daemon=True)
t.start()
# Stop after 10 seconds (in a real tool, this would be a signal handler)
import time
time.sleep(10)
stop.set()
t.join()
Output:
modified: /project/data.json
Watcher stopped.
stop_event accepts any object with a .is_set() method — a threading.Event works perfectly. This pattern is essential when the watcher runs in a background thread and needs to be shut down cleanly when the main process exits or receives a signal.
Filtering Changes
In real projects you usually care about specific file types. watchfiles has a built-in filtering mechanism via the watch_filter parameter that accepts a callable returning True to include a change or False to ignore it.
# watch_filter.py
from watchfiles import watch, Change
def python_files_only(change: Change, path: str) -> bool:
"""Only watch .py files, ignore __pycache__ and .pyc."""
return path.endswith(".py") and "__pycache__" not in path
for changes in watch("./src", watch_filter=python_files_only):
for change_type, path in changes:
print(f"{change_type.name}: {path}")
Output:
modified: /project/src/main.py
modified: /project/src/utils.py
The filter receives the Change type and the full file path as a string. This means you can filter on both: ignore deleted events for .log files but react to all events for .py files, for example. Returning False drops the event entirely — it never reaches the outer loop.
Built-In Filter Classes
watchfiles also ships two ready-made filter classes for the most common cases.
# watch_builtin_filters.py
from watchfiles import watch
from watchfiles.filters import PythonFilter, DefaultFilter
# PythonFilter: watches only .py files, skips __pycache__ and test files
for changes in watch("./app", watch_filter=PythonFilter()):
for _, path in changes:
print(f"Python file changed: {path}")
# DefaultFilter: ignores .git/, __pycache__/, .pytest_cache/, temp files
# This is the default if you don't set watch_filter at all
for changes in watch(".", watch_filter=DefaultFilter()):
for change_type, path in changes:
print(f"{change_type.name}: {path}")
Output:
Python file changed: /project/app/models.py
modified: /project/README.md
PythonFilter is the right choice for hot-reload tools that only care about source code. DefaultFilter is a sensible default for general file watching — it silences the noise from version control, caches, and editor temp files that would otherwise fire constantly as you work.
Async Watching with awatch()
For async code — FastAPI, aiohttp, asyncio scripts — use awatch(). It is the async equivalent of watch() and fits naturally into async for loops without blocking the event loop.
# awatch_basic.py
import asyncio
from watchfiles import awatch
async def watch_configs():
print("Watching config directory...")
async for changes in awatch("./config"):
for change_type, path in changes:
print(f"Config changed: {change_type.name} -- {path}")
await reload_config(path)
async def reload_config(path: str):
"""Simulate an async config reload."""
print(f" Reloading from {path}...")
await asyncio.sleep(0.1) # simulate async I/O
print(f" Done.")
asyncio.run(watch_configs())
Output (after editing config/settings.yml):
Watching config directory...
Config changed: modified -- /project/config/settings.yml
Reloading from /project/config/settings.yml...
Done.
awatch() accepts the same parameters as watch(), including watch_filter and stop_event. The stop event for async is an asyncio.Event instead of a threading.Event. Inside the loop body, you can await any coroutine — reading a file, reloading a cache, posting a webhook — without blocking other tasks in the event loop.
Debouncing
Editors often write a file in multiple rapid steps — a save triggers a rename of the temp file, then a write, then a permission update. Without debouncing, you would get 3-5 events for a single conceptual change. watchfiles debounces by default, batching changes that arrive within 100 milliseconds into one set.
# watch_debounce.py
from watchfiles import watch
# Default: debounce_ms=100 (100 ms batching window)
for changes in watch("./src"):
print(f"Got {len(changes)} change(s) in one batch")
for change_type, path in changes:
print(f" {change_type.name}: {path}")
# Reduce to 50ms for faster reaction (at the cost of more batches)
for changes in watch("./src", debounce=50):
print(f"Got {len(changes)} change(s)")
# Increase to 500ms to group many rapid changes together
for changes in watch("./src", debounce=500):
print(f"Got {len(changes)} change(s)")
Output (after one ctrl+S save in VS Code):
Got 3 change(s) in one batch
modified: /project/src/main.py
modified: /project/src/main.py
modified: /project/src/main.py
Even at 100ms debounce, a single save can produce multiple modified events for the same file if the editor writes in stages. The deduplication responsibility is still yours — filter the set to unique paths before acting. A simple paths = {path for _, path in changes} gives you the distinct files without worrying about how many events each generated.
Real-Life Example: YAML Config Auto-Reloader
This example builds a service that loads a YAML configuration file on startup and automatically reloads it whenever the file changes — without restarting the process. This is the pattern used by web servers, task queues, and daemons that need live config updates.
# config_reloader.py
import asyncio
import time
from pathlib import Path
import yaml # pip install pyyaml
from watchfiles import awatch
from watchfiles.filters import DefaultFilter
CONFIG_PATH = Path("config/settings.yml")
# --- Simulated app state ---
class AppConfig:
def __init__(self):
self.data: dict = {}
self.loaded_at: float = 0.0
def load(self, path: Path) -> None:
"""Load config from YAML file."""
try:
with path.open() as f:
self.data = yaml.safe_load(f) or {}
self.loaded_at = time.time()
print(f"[config] Loaded {len(self.data)} keys from {path.name}")
except FileNotFoundError:
print(f"[config] WARNING: {path} not found -- using empty config")
self.data = {}
except yaml.YAMLError as exc:
print(f"[config] ERROR: Invalid YAML -- {exc}")
config = AppConfig()
# --- Watcher coroutine ---
async def watch_config(stop: asyncio.Event) -> None:
"""Watch settings.yml and reload on every change."""
print(f"[watcher] Watching {CONFIG_PATH} for changes...")
async for changes in awatch(
CONFIG_PATH.parent,
watch_filter=DefaultFilter(),
stop_event=stop,
):
changed_paths = {path for _, path in changes}
if str(CONFIG_PATH.resolve()) in {str(Path(p).resolve()) for p in changed_paths}:
print(f"[watcher] {CONFIG_PATH.name} changed -- reloading...")
config.load(CONFIG_PATH)
print(f"[watcher] debug_mode = {config.data.get('debug_mode', False)}")
# --- Main application ---
async def main() -> None:
# Create a sample config if it doesn't exist
CONFIG_PATH.parent.mkdir(exist_ok=True)
if not CONFIG_PATH.exists():
CONFIG_PATH.write_text("debug_mode: false\nlog_level: info\nmax_workers: 4\n")
# Load initial config
config.load(CONFIG_PATH)
stop = asyncio.Event()
watcher_task = asyncio.create_task(watch_config(stop))
print("[app] Running. Edit config/settings.yml to see a live reload.")
print("[app] Press Ctrl+C to stop.")
try:
await asyncio.sleep(60) # simulate app running
except asyncio.CancelledError:
pass
finally:
stop.set()
await watcher_task
print("[app] Shutdown complete.")
asyncio.run(main())
Output (after editing settings.yml to set debug_mode: true):
[config] Loaded 3 keys from settings.yml
[app] Running. Edit config/settings.yml to see a live reload.
[app] Press Ctrl+C to stop.
[watcher] Watching config/settings.yml for changes...
[watcher] settings.yml changed -- reloading...
[config] Loaded 3 keys from settings.yml
[watcher] debug_mode = True
The key design decisions: the watcher task runs as a background asyncio task alongside the main application coroutine. The asyncio.Event provides a clean shutdown path — when the main coroutine exits, it sets stop, which causes awatch() to exit its loop. To extend this into a production pattern, replace the AppConfig.load() method with your actual config parsing, and replace asyncio.sleep(60) with whatever your application actually does.
Frequently Asked Questions
What is the difference between watchfiles and watchdog?
watchdog is a well-established Python library that wraps OS file-system APIs using a thread-based observer pattern with event handler classes. watchfiles is newer, uses a Rust backend for lower overhead, and exposes a simpler iterator API instead of the observer/handler pattern. Both use native OS events on supported platforms. For async Python code, watchfiles integrates more naturally via awatch(). For projects already using watchdog with complex handler hierarchies, the migration cost may not be worth it — but for new projects, watchfiles requires less boilerplate.
Does watchfiles watch subdirectories recursively?
Yes, by default. Passing "./src" to watch() watches the entire tree under src/, including all subdirectories and their files. There is no recursive=True flag needed — it is always on. To limit to a single directory without recursion, you would need a custom watch_filter that checks the path depth.
Does watchfiles follow symlinks?
watchfiles watches the resolved path of symlinked directories. If ./config is a symlink pointing to /etc/myapp/config, events arrive for changes to the real files at /etc/myapp/config/. However, the paths in the change set will be the real paths, not the symlink path. This can cause a mismatch if your code compares paths to the symlink location — always resolve paths with Path.resolve() before comparing.
How does watchfiles perform on large directory trees?
Since watchfiles uses OS native events rather than polling, adding more files to a watched directory does not increase CPU usage during idle periods. The OS kernel registers watches at the inode level and only wakes the process when an event occurs. In practice, watching a 10,000-file repository behaves the same as watching a 100-file project at rest. The startup cost of registering all watches grows with the number of directories, but after that the overhead is near-zero.
Can I react to changes and run code in the same process vs a subprocess?
Both patterns work. For config reloading or cache invalidation, running the handler code in the same process (as in the real-life example above) is the right choice — no inter-process communication needed, and state is shared directly. For hot-reloading a web server or long-running service, spawning a subprocess and killing/restarting it on changes is safer — it guarantees a clean state with no leftover globals. Tools like uvicorn use watchfiles in process-restart mode: they watch the source tree and os.execv() themselves when Python files change.
Does watchfiles work on Windows?
Yes. On Windows, watchfiles uses the ReadDirectoryChangesW API, which is the same native event source used by most Windows file watchers. The Python API is identical across platforms — you write the same watch() or awatch() code and it works on Linux, macOS, and Windows without any conditional imports or platform-specific configuration.
Conclusion
Python’s watchfiles library makes file watching a one-import, one-loop operation. This article covered the synchronous watch() iterator, stopping watchers cleanly with stop_event, filtering by file type with custom functions and built-in PythonFilter / DefaultFilter classes, the async awatch() function for event-loop-friendly code, debounce configuration, and a complete async YAML config reloader. The library’s Rust backend means near-zero CPU cost at rest, making it appropriate for tools that run in the background for hours or days.
To extend the config reloader: add a schema validation step using Pydantic after each reload, send a notification to a Slack channel when the config changes using httpx, or combine watchfiles with rich‘s live display to show a real-time feed of changes in a terminal dashboard. Each of those extensions is under 20 additional lines.
For the full API reference, including watch_filter signatures, all constructor parameters, and advanced usage with rust_timeout and yield_on_timeout, see the official documentation at watchfiles.helpmanual.io.