Plombery 实时日志深度剖析get_logger() 背后的 contextvars 魔法与 WebSocket 推送原理【免费下载链接】plomberyPython task scheduler with a user-friendly web UI项目地址: https://gitcode.com/gh_mirrors/pl/plomberyPlombery 是一个用 Python 编写的任务调度器自带用户友好的 Web 界面。它的最大亮点之一是你在浏览器里看到的 Pipeline 运行日志能像终端一样实时刷新——背后是get_logger()函数、Python 的contextvars模块和一条 WebSocket 推送链路共同完成的魔法。本文将带你从日志诞生到抵达屏幕完整走一遍这条实时日志流。为什么 Web 面板里的日志是活的 ️大多数调度系统只把日志写进文件你打开页面看到的永远是上一次刷新的快照。Plombery 不同日志一边落盘、一边通过 Socket.IO 按运行run维度广播前端订阅logs.{run_id}事件后逐条追加渲染。整体链路可以概括为三步任务函数调用get_logger()获取一个知道自己属于哪次运行、哪个任务的 Logger两条 Handler 同时工作FileHandler把 JSON 日志写入本次运行的日志文件QueueHandler把日志塞进队列后台监听线程从队列取出日志经WebSocketHandler通过 Socket.IO 推送到logs.{run_id}事件。get_logger() 背后的 contextvars 魔法打开源码src/plombery/logger/__init__.pyget_logger()的第一件事是pipeline pipeline_context.get() task task_context.get(None) pipeline_run run_context.get()它没有接收任何参数却知道当前正在执行哪条 Pipeline、哪次运行、哪个任务。秘密就在src/plombery/pipeline/context.py中定义的全局ContextVarpipeline_context: ContextVar[Pipeline] ContextVar(pipeline) task_context: ContextVar[Task] ContextVar(task) run_context: ContextVar[PipelineRun] ContextVar(run)三个上下文是如何被绑上的 Pipeline 与 Run调度器在真正执行 Pipeline 前src/plombery/orchestrator/executor.py中的run()函数先创建运行记录然后pipeline_context.set(pipeline)和run_context.set(pipeline_run)执行结束后用保存的 token 精确重置Task你用task装饰器定义的每个任务函数都会被src/plombery/pipeline/__init__.py中的包装器在调用前后task_context.set(task)/task_context.reset(token)包裹。这就是contextvars的价值它让当前运行上下文像环境变量一样随调用栈流动却不会在全局变量里打架——即使多个 Pipeline 并发运行各自的日志也天然隔离。同步任务函数通过asyncio.to_thread执行上下文同样能传播到线程里。一条日志两个去处JSON 落盘 WebSocket 推送get_logger()为每次运行创建独立命名的 Loggerplombery.{run_id}任务级还会追加-{task_id}挂载两个 HandlerFileHandlerJsonFormatter日志以 JSON 行JSONL写入该次运行专属的日志文件路径由get_logs_filename()src/plombery/orchestrator/data_storage.py生成保证历次运行互不覆盖。JsonFormattersrc/plombery/logger/formatter.py还会自动附加pipeline、task、UTC 时间戳和异常堆栈前端正是靠这些字段做筛选和着色QueueHandler日志不直接发网络请求而是先进入一个无界队列。队列 监听线程不阻塞业务的推送设计src/plombery/logger/web_socket_handler.py实现了经典的异步解耦套路queue_handler QueueHandler(_logs_queue) handler WebSocketHandler() _listener QueueListener(_logs_queue, handler) _listener.start()QueueListener是一条独立后台线程任务代码只负责把日志扔进队列非阻塞由监听线程统一取出调用WebSocketHandler.emit()执行sio.emit(flogs.{record.run_id}, record.message)。这保证了高频打日志的任务函数不会被网络 IO 拖慢。前端如何接住这些日志WebSocket 服务端是标准 Socket.IOsrc/plombery/websocket.py中创建AsyncServer挂载在/ws路径。前端在frontend/src/socket.ts建立连接后frontend/src/components/LogViewer.tsx组件订阅对应运行的事件socket.on(\logs.${run.id}, onWsMessage)收到消息后解析 JSON追加到本地缓存并触发表格重渲染历史日志则通过 REST 接口一次性拉取src/plombery/api/routers/runs.py的/{run_id}/logs端点返回application/jsonl实现先出历史、再接实时的无缝体验页面上的任务筛选、日志级别筛选、自动滚动到底部都基于日志 JSON 中的task、level字段完成。核心源码文件清单 文件职责src/plombery/logger/__init__.pyget_logger()/close_logger()日志装配核心src/plombery/pipeline/context.pypipeline_context/task_context/run_context三个 ContextVarsrc/plombery/pipeline/__init__.pytask装饰器绑定任务上下文src/plombery/orchestrator/executor.py运行调度器设置 Pipeline/Run 上下文src/plombery/logger/formatter.pyJSON 行格式化器src/plombery/logger/web_socket_handler.py队列 监听线程 WebSocket 推送src/plombery/websocket.pySocket.IO ASGI 服务端frontend/src/components/LogViewer.tsx实时日志查看器组件小结Plombery 的实时日志看似简单实则是一套教科书级的组合拳contextvars解决日志属于谁JSONL 落盘 队列解耦解决日志存到哪、怎么不阻塞Socket.IO 按 run_id 分事件解决日志如何精准送达。理解了这三层你也能在自己的 Python Web 应用里复刻一条实时日志流。【免费下载链接】plomberyPython task scheduler with a user-friendly web UI项目地址: https://gitcode.com/gh_mirrors/pl/plombery创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考