Python中如何实现异步的文件系统监控_使用watchfiles库替代同步watchdog-Python教程-PHP中文网

watchfiles 更适合异步场景，因其原生基于 asyncio 设计，直接返回异步迭代器，无需线程模拟或 executor 封装；而 watchdog 依赖同步事件循环和轮询线程，强行接入 async 代码易引发未 await 警告、异常丢失或阻塞问题。

python中如何实现异步的文件系统监控_使用watchfiles库替代同步watchdog

watchfiles 为什么比 watchdog 更适合异步场景

因为 watchfiles 从设计上就是纯异步的 —— 它不依赖线程模拟异步，也不需要你在事件回调里手动 await 什么，而是直接返回一个异步迭代器。而 watchdog 的核心是同步事件循环 + 线程轮询，想在 async def 函数里用它，只能靠 loop.run_in_executor 包一层，既麻烦又容易漏掉异常、阻塞事件循环。

常见错误现象：RuntimeWarning: coroutine 'AsyncIOHandler.on_modified' was never awaited —— 这说明你试图把异步 handler 直接塞进 watchdog 的同步回调链路里，Python 发现它没被 await 就丢弃了。

watchfiles 底层用的是 inotify（Linux）、FSEvents（macOS）或 ReadDirectoryChangesW（Windows），和 watchdog 类似，但封装层完全跑在 asyncio 上
它不提供“监听器注册/取消”这种面向对象接口，而是用 awatch() 返回一个 AsyncIterator[Change]，符合现代 async Python 的使用直觉
没有后台线程，不会意外干扰你的 asyncio.run() 或 uvicorn 生命周期

如何正确启动一个异步文件监控任务

最简可用模式就是用 async for 遍历 awatch() 返回的异步迭代器。注意：它会一直阻塞等待变更，所以通常要配合 asyncio.create_task() 或放在主协程里跑。

import asyncio
from watchfiles import awatch
<p>async def main():
async for changes in awatch('./src', watch<em>filter=lambda </em>, path: path.endswith('.py')):
for change_type, file_path in changes:
print(f'{change_type}: {file_path}')</p><h1>这里可以 await 其他协程，比如 reload module、触发测试等</h1><p>asyncio.run(main())</p>

awatch() 默认递归监听子目录；加 recursive=False 可关闭
watch_filter 是个同步函数，不能是 async def —— 它在 C 层过滤路径，必须快，不能 await
如果监听路径不存在，awatch() 会立即抛出 FileNotFoundError，不是静默跳过
不要在 async for 循环里做耗时同步操作（如 time.sleep(5)），会卡住整个监听流

处理 change 类型与路径拼接的坑

changes 是一个 set，每个元素是 (Change, str) 元组，其中 Change 是枚举值：Change.added、Change.modified、Change.deleted。注意它不保证顺序，也不合并重复事件（例如快速保存两次，可能收到两个 modified）。

Python 3.14.2

Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本，属于3.14系列的第二个维护更新。该版本包含了18项修复，重点解决了多进程、数据类及正则表达式等模块的回归问题，并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式（移除GIL）正式获得官方支持，是Python发展的重要里程碑。

下载

常见错误现象：OSError: [Errno 2] No such file or directory: '/path/to/deleted.txt' —— 因为 deleted 事件发生后，文件已不在磁盘，你还去 open() 它。

对 Change.deleted，只应做清理逻辑（如从缓存移除路径），不要尝试读取文件内容
file_path 是相对监听根目录的路径，不是绝对路径；需用 Path('./src') / file_path 拼成绝对路径再操作
Windows 下路径分隔符是 \，但 watchfiles 统一返回 / 风格（兼容 POSIX），无需额外 normalize
重命名（move）会被拆成 deleted + added 两个事件，无法原子识别 —— 这是底层 API 限制，不是库的问题

与 FastAPI/Uvicorn 集成时的生命周期管理

如果你在 FastAPI 启动时开启监听，得确保它能随服务一起 shutdown，否则进程退出时可能卡在 awatch() 的等待中。

Uvicorn 不支持直接 await 协程作为 startup hook，所以要用 asyncio.create_task() 并存引用，再在 shutdown 时 task.cancel()。

from fastapi import FastAPI
import asyncio
from watchfiles import awatch
<p>app = FastAPI()
_watch_task = None</p><p>@app.on_event("startup")
async def start_watcher():
global _watch_task
_watch_task = asyncio.create_task(watch_loop())</p><p>@app.on_event("shutdown")
async def stop_watcher():
if _watch_task and not _watch_task.done():
_watch_task.cancel()
try:
await _watch_task
except asyncio.CancelledError:
pass</p><p>async def watch_loop():
try:
async for changes in awatch('./config'):
for change, path in changes:
if change == Change.modified:</p><h1>reload config</h1><pre class="brush:python;toolbar:false;">                pass
except asyncio.CancelledError:
    raise
except Exception as e:
    print(f"Watcher error: {e}")