1.watchfiles #
本节介绍 watchfiles 的基本定位。watchfiles 是一个高效的文件系统监视库(由 UV/Rye 作者开发),可在项目文件变更时快速触发回调或重启子进程。它适用于本地开发自动重载、脚手架工具、构建系统监听等场景。
2.核心特性 #
本节概述 watchfiles 的主要优势,帮助你判断何时选择它作为文件监控与开发时自动重载的方案。
- 跨平台支持:Windows/macOS/Linux 全兼容
- 高性能:使用 Rust 底层实现,比纯 Python 方案快 5-10 倍
- 精确检测:支持文件内容变更检测(不只是修改时间)
- 最小化重启:智能合并连续变更事件
3.安装方法 #
本节展示在 Windows 环境下的安装方式。建议先升级 pip,再安装 watchfiles。
REM 升级 pip,避免旧版本导致安装失败
py -m pip install --upgrade pip
REM 安装 watchfiles(如需国内镜像可自行追加 -i 参数)
py -m pip install watchfiles4.基础用法 #
本节给出最常见的两种用法:作为命令行工具直接监听并重启进程,以及在代码中以编程方式集成。
4.1. 作为 CLI 工具使用 #
当你想要在文件变更时重启某个命令(如运行应用)时,可使用 CLI 形式。下例监听当前目录,文件变化时重启 py app.py。
REM 监听当前目录,并在变更时运行/重启目标命令
py -m watchfiles "py app.py" .常用参数(可与上述命令组合使用):
--verbose显示详细变更信息--ignore-paths忽略特定路径--extensions只监视特定扩展名文件
4.2. 编程方式使用 #
当你需要在应用内部获取更细粒度的变更事件,或将监听与自定义逻辑/子进程管控结合时,推荐使用编程方式集成。
4.2.1. 基本监视循环 #
该示例演示如何以最简单的方式监听目录变更,并打印所有变更事件集合。
# 从 watchfiles 导入 watch 函数
from watchfiles import watch
# 定义脚本主函数
def main():
# 对指定目录进行监听(此处为 ./project_dir)
for changes in watch('./project_dir'):
# 打印捕获到的变更事件集合(集合内为二元组:变更类型、文件路径)
print(f'检测到变更: {changes}')
# 脚本入口
if __name__ == '__main__':
# 调用主函数
main()4.2.2. 与子进程集成 #
该示例展示如何在文件变更时自动重启一个子进程(如运行你的应用脚本),并在每次准备重启时执行回调。
# 导入 run_process 以便在文件变更时重启子进程
from watchfiles import run_process
# 导入 sys 以演示可选的命令行参数透传(本例未强制使用)
import sys
# 定义变更回调函数(在每次重启前被调用)
def on_change(changes):
# 打印即将触发重启的信息
print(f'准备重启,检测到变更: {changes}')
# 脚本入口
if __name__ == '__main__':
# 监听 ./project_dir 目录,在变更时以子进程方式运行 `py app.py`
# 指定监听目录(当该目录内文件发生变化时触发)
run_process(
# 要监听的路径
'./project_dir',
# 启动的目标可执行名称(Windows 下建议使用 'py' 以调用 Python)
target='py',
# 传递给目标程序的参数列表(此处运行 app.py)
args=['app.py'],
# 每次准备重启前执行的回调函数
callback=on_change
)5.高级功能 #
当你需要更精确的控制(如忽略特定目录/文件、只接收某些扩展名、处理不同事件类型、异步监听等),可以使用以下高级特性。
5.1. 过滤文件 #
通过自定义 watch_filter(继承 DefaultFilter)来灵活地过滤不关心的变更,如临时文件、测试目录等。
# 导入 watch、默认过滤器基类
from watchfiles import watch, DefaultFilter
# 导入 os 以实现跨平台路径判断
import os
# 自定义过滤器,继承 DefaultFilter 并覆写 __call__
class MyFilter(DefaultFilter):
# 每个事件都会调用该函数,返回 True 表示该路径/事件应被保留
def __call__(self, change, path):
# 构造跨平台的测试目录片段(Windows 使用反斜杠,UNIX 使用斜杠)
tests_segment = f'{os.sep}tests{os.sep}'
# 组合默认过滤器逻辑 + 自定义忽略规则
# 返回布尔表达式,逐项合并判断
return (
# 先沿用默认过滤规则(忽略常见无关路径/文件)
super().__call__(change, path)
# 额外忽略 .tmp 临时文件
and not path.endswith('.tmp')
# 额外忽略 tests 目录下的变更
and tests_segment not in path
)
# 脚本入口
def main():
# 监听当前目录,应用自定义过滤器
for changes in watch('.', watch_filter=MyFilter()):
# 打印过滤后的变更事件
print(changes)
# 主入口
if __name__ == '__main__':
# 调用主函数
main()5.2. 自定义事件处理 #
当需要区分新增、修改、删除等不同事件类型时,可以使用 Change 枚举进行精细化处理。
# 导入 Change 枚举与 watch 函数
from watchfiles import Change, watch
# 定义主函数
def main():
# 监听当前目录,逐事件处理
for changes in watch('.'):
# 遍历每条事件(一条事件是二元组:事件类型、文件路径)
for change_type, path in changes:
# 判断是否为新增文件
if change_type == Change.added:
# 打印新增文件路径
print(f'新增文件: {path}')
# 判断是否为修改文件
elif change_type == Change.modified:
# 打印修改文件路径
print(f'修改文件: {path}')
# 判断是否为删除文件
elif change_type == Change.deleted:
# 打印删除文件路径
print(f'删除文件: {path}')
# 脚本入口
if __name__ == '__main__':
# 调用主函数
main()5.3. 与异步框架集成 #
如果你的应用使用 asyncio,可以用 awatch 在异步循环中监听变更,不会阻塞主事件循环。
# 导入 asyncio 以使用异步事件循环
import asyncio
# 从 watchfiles 导入异步监听函数 awatch
from watchfiles import awatch
# 定义异步主函数
async def main():
# 使用异步 for 循环持续监听
async for changes in awatch('.'):
# 打印捕获到的变更事件集合
print(f'异步检测到变更: {changes}')
# 脚本入口
if __name__ == '__main__':
# 运行异步主函数
asyncio.run(main())6.性能对比 #
本节提供与常见替代品的对比数据(示例值,仅供参考)。实际性能受磁盘、文件数量与变更频率影响。
| 工具 | 首次扫描 | 增量更新 | 内存占用 |
|---|---|---|---|
| watchfiles | ~50ms | ~5ms | ~10MB |
| watchdog | ~200ms | ~20ms | ~30MB |
| pyinotify | ~150ms | ~15ms | ~25MB |
7.实际应用示例 #
本节包含一个典型的“开发服务器自动重启”场景:当代码有变更时,自动重启运行中的服务(以 uvicorn 为例)。
7.1. 开发服务器自动重启 #
示例脚本 autoreload.py:监听当前目录,在变更时重启 uvicorn 运行的 ASGI 应用。
# 从 watchfiles 导入 run_process 以在变更时重启子进程
from watchfiles import run_process
# 导入 sys(可用于扩展参数透传,本例未强制使用)
import sys
# 定义回调函数:在每次重启前被调用
def on_reload(changes):
# 打印即将重启的信息
print(f'文件变更,重启服务: {changes}')
# 脚本入口
if __name__ == '__main__':
# 监听当前目录,变化时重启 uvicorn(请确保已安装 uvicorn)
# 开始调用 run_process,监听目录并托管子进程
run_process(
# 要监听的路径(当前目录)
'.',
# 子进程入口(uvicorn 可执行名)
target='uvicorn',
# 传递给 uvicorn 的参数(模块:应用 ,以及端口)
args=['main:app', '--port', '8000'],
# 重启前的回调(打印变更)
callback=on_reload
)运行(Windows 命令提示符):
REM 运行自动重载脚本
py autoreload.py8.最佳实践 #
本节列出在实际项目中常见的配置技巧,帮助你减少无效触发、提升开发体验与性能。
8.1. 忽略虚拟环境 #
忽略虚拟环境目录可避免第三方包更新导致的无意义重启。
# 从 watchfiles 导入 watch 函数
from watchfiles import watch
# 定义主函数
def main():
# 监听当前目录,忽略常见虚拟环境目录
for changes in watch('.', ignore_paths=['.venv', 'venv', 'env']):
# 打印检测到的变更事件
print(changes)
# 脚本入口
if __name__ == '__main__':
# 调用主函数
main()8.2. 限制文件类型 #
仅监听特定扩展名(如 .py)能避免无关文件引发重启,减少噪音与开销。
# 导入 watch 函数
from watchfiles import watch
# 定义主函数
def main():
# 监听当前目录,仅保留以 .py 结尾的文件事件
for changes in watch('.', watch_filter=lambda change, path: path.endswith('.py')):
# 打印过滤后的变更事件
print(changes)
# 脚本入口
if __name__ == '__main__':
# 调用主函数
main()8.3. 防抖处理 #
在频繁写入的场景(例如编译产物、批量保存)下,可以加一层简单的“防抖”逻辑,避免过于频繁地重启服务。
# 从 watchfiles 导入 watch 函数
from watchfiles import watch
# 导入 time 以实现简单的时间窗
import time
# 定义一个示例的重启函数(实际中替换为你的业务逻辑)
def restart_server():
# 打印重启提示
print('重启服务...')
# 定义主函数
def main():
# 记录上一次重启时间戳(初始化为 0)
last_restart = 0.0
# 监听当前目录
for changes in watch('.'):
# 获取当前时间
now = time.time()
# 若距离上次重启超过 2 秒,则执行重启
if now - last_restart > 2:
# 调用重启函数
restart_server()
# 更新上次重启时间戳
last_restart = now
# 脚本入口
if __name__ == '__main__':
# 调用主函数
main()watchfiles 特别适合与 UV(uv run/uvx)等开发工具配合使用,能为 Python 项目提供高效可靠的开发时自动重载功能。