FastAPI收到请求时如何实时更新UI?
FastAPI收到请求时如何实时更新UI?
看起来你是想把FastAPI收到请求的日志实时同步到Gradio的UI上对吧?我之前也遇到过类似的问题,核心难点在于FastAPI和Gradio运行在不同的线程里,直接修改共享变量UI不会自动刷新,这里给你一套可行的解决方案:
首先先把你那截断的代码补全并优化一下,先定义好共享的日志存储类,确保线程安全:
import os import gradio as gr from fastapi import FastAPI, Request import uvicorn import threading from typing import List from datetime import datetime import time api = FastAPI() # 共享日志存储,加锁保证线程安全 class Log(): def __init__(self): self._logs: List[str] = [] self.logstr = "" self.lock = threading.Lock() def log_message(self, msg: str): with self.lock: timestamp = datetime.now().strftime("%H:%M:%S") self._logs.append(f"[{timestamp}] {msg}") # 限制日志条数,避免UI太长 if len(self._logs) > 50: self._logs.pop(0) self.logstr = "\n".join(self._logs) log_store = Log()
接下来写FastAPI的接口,收到请求时直接调用日志类的方法记录:
# FastAPI接口示例 @api.post("/webhook") async def webhook(request: Request): data = await request.json() log_store.log_message(f"收到POST请求:{str(data)}") return {"status": "success"} @api.get("/test") async def test(): log_store.log_message("收到GET测试请求") return {"status": "success", "msg": "测试完成"}
然后是Gradio的UI部分,关键是用定时轮询的方式去拉取最新的日志,实现实时更新:
# Gradio实时日志UI def refresh_logs(): while True: yield log_store.logstr time.sleep(0.5) with gr.Blocks() as demo: gr.Markdown("# FastAPI请求实时日志") log_display = gr.Textbox(label="请求记录", lines=15, interactive=False, placeholder="等待请求...") # 每隔0.5秒自动刷新日志 log_display.change(fn=refresh_logs, outputs=log_display, every=0.5)
最后把两个服务放在不同线程里启动,避免互相阻塞:
# 启动FastAPI线程 def run_fastapi(): uvicorn.run(api, host="0.0.0.0", port=8000) if __name__ == "__main__": threading.Thread(target=run_fastapi, daemon=True).start() demo.launch(server_name="0.0.0.0", server_port=7860, share=False)
原理说明
- 共享的
log_store是跨线程的存储,加了锁之后不管多少个FastAPI请求同时写入,都不会出现日志错乱的问题。 - Gradio这边用生成器函数
refresh_logs定时(每0.5秒)返回最新的日志字符串,通过every参数触发UI自动更新,这样就能实时看到FastAPI收到的请求日志了。
如果你觉得0.5秒的间隔不合适,可以自己调整,比如改成0.2秒会更灵敏,但也别太频繁,不然会占用过多资源。
备注:内容来源于stack exchange,提问作者Dimitrios Desyllas
相关产品推荐
相关产品推荐

