python watchdog 配置文件热更新

一、Watchdog示例

二、aiohttp服务配置热更新

在同事的golang代码中学习到了config.json热更新的功能，这里自己也学习了一下python写web服务的时候怎么来实现配置的热更新。主要是利用Watchdog这个第三方python库，来监控文件系统的改变，从而实现不用重启代码热更新的功能。

一、Watchdog示例

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandlerclass HotUpdateEventHandler(FileSystemEventHandler):def on_created(self, event):if event.is_directory:returnprint(f'File created: {event.src_path}')def on_modified(self, event):if event.is_directory:returnprint(f'File modified: {event.src_path}')if __name__ == "__main":path = "/root"  # 检测根目录下文件和目录的改变# 处理器event_handler = HotUpdateEventHandler()# 监控器observer = Observer()observer.schedule(event_handler, path, recursive=False)observer.start()try:while True:# 其他的功能逻辑do_something()except KeyboardInterrupt:observer.stop()observer.join()

流程：

1、定义事件处理器(文件的修改、删除等对应逻辑)

2、定义监控器

3、监控器中执行调度，传入事件处理器、监控路径和是否递归监控

重要的两个class

watchdog.observers.Obsever

watchdog.events.FileSystemEventHandler

Obsever发起监控线程，FileSystemEventHandler实现具体的文件系统变化的逻辑

在实现具体的功能的时候，需要自己重写FileSystemEventHandler中的方法

主要常用的就是以上的方法

其中event具有的属性如下

二、aiohttp服务配置热更新

import json
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import aiohttp
from aiohttp import web
import os
config_path = "./test.json"
config = json.load(open(config_path))def reload_config():"""热重载配置"""global configtry:new_config = json.load(open(config_path))print('Config reloaded successfully.')config = new_configexcept Exception as e:print(f'Failed to reload config: {e}')class ConfigReloadEventHandler(FileSystemEventHandler):def on_modified(self, event):if event.src_path == config_path:print(f"Config file modified: {event.src_path}. Reloading...")reload_config()async def post(request:web.Request):req = await request.json()send_data = configreturn web.json_response(send_data)async def main():app = web.Application()app.add_routes([web.post('/nlp', post)])return app# 使用方法
if __name__ == "__main__":config_event_handler = ConfigReloadEventHandler()path, filename = os.path.split(config_path)observer = Observer()observer.schedule(config_event_handler,path,recursive=False)observer.start()try:local_ip = "0.0.0.0"port = "6666"web.run_app(main(), port=port)except KeyboardInterrupt:observer.stop()observer.join()

使用aiohttp启动一个web 服务，同时热更新test.json配置文件，作为全局变量config在web服务中使用