主页 / coding / 问题 / 79598793
Accepted
Dimitrios Desyllas
Asked: 2025-04-30 00:08:50 +0800 CST

当我在 FastAPI 上收到请求时，如何实时更新 Ui？

  • 772

我有这个简单的脚本：

import os
import gradio as gr
from fastapi import FastAPI, Request
import uvicorn
import threading
from typing import List
from datetime import datetime

api = FastAPI()

# Shared logs

class Log():
    def __init__(self):
        self._logs: List[str] = []
        self.logstr=""
    
    def log_message(self,msg: str):
        timestamp = datetime.now().strftime("%H:%M:%S")
        self._logs.append(f"[{timestamp}] {msg}")
        self.logstr="\n".join(self._logs)


log = Log()
log_state=gr.State(log)

# === FastAPI Setup ===


@api.post("/log")
async def receive_log(request: Request):
    data = await request.body()
    msg = f"API received: {data}"
    log.log_message(msg)
    gr.update(value=log.logstr)
    return {"status": "logged", "message": msg}

def run_api():
    api_port = int(os.environ.get("API_PORT", 8000))
    uvicorn.run(api, host="0.0.0.0", port=api_port)

# === Gradio UI ===

with gr.Blocks() as ui:
    gr.Markdown("## 📝 Incoming HTTP Requests")
    log_box = gr.Textbox(label="Logs", inputs=log_state, lines=20)
    
    # Trigger the refresh when the log state is updated


def run_gradio():
    gradio_port = int(os.environ.get("GRADIO_PORT", 7860))
    ui.launch(server_port=gradio_port)

# === Start Both ===
if __name__ == "__main__":
    threading.Thread(target=run_api, daemon=True).start()
    run_gradio()

我尝试实现的目标是让 FastAPI 监听一个端口和一个实时显示传入请求的面板：

POST /log -> FastAPI ->common_log -> Gradio

但是我无法在 FastAPI 中更改接收传入请求时的内容Textbox。我该怎么做？

python

2 个回答

  1. Best Answer

    对我有用的唯一方法是

    def function():
    return log.logstr

Textbox(value=function, ..., every=1)

    它function每 1 秒运行一次，并function返回日志中的当前内容。

    文档：Gradio 文本框

    完整代码：

    import os
import gradio as gr
from fastapi import FastAPI, Request
import uvicorn
import threading
from typing import List
from datetime import datetime

api = FastAPI()

# Shared logs

class Log():
    def __init__(self):
        self._logs: List[str] = []
        self.logstr=""

    def log_message(self,msg: str):
        timestamp = datetime.now().strftime("%H:%M:%S")
        self._logs.append(f"[{timestamp}] {msg}")
        self.logstr="\n".join(self._logs)

log = Log()
log_state = gr.State(log)

# === FastAPI Setup ===

@api.post("/log")
async def receive_log(request: Request):
    data = await request.body()
    msg = f"API received: {data}"
    log.log_message(msg)
    gr.update(value=log.logstr)
    #print('data:', data)
    return {"status": "logged", "message": msg}

def run_api():
    #print('run FastAPI')
    api_port = int(os.environ.get("API_PORT", 8000))
    uvicorn.run(api, host="0.0.0.0", port=api_port)

# === Gradio UI ===

def get_logs():
    #print('run: get_logs')
    return log.logstr

with gr.Blocks() as ui:
    gr.Markdown("## 📝 Incoming HTTP Requests")
    log_box = gr.Textbox(label="Logs", value=get_logs, lines=20, every=1)
    # Trigger the refresh when the log state is updated

def run_gradio():
    gradio_port = int(os.environ.get("GRADIO_PORT", 7860))
    ui.launch(server_port=gradio_port)

# === Start Both ===

if __name__ == "__main__":
    threading.Thread(target=run_api, daemon=True).start()
    run_gradio()
    • 1

  2. 我采用了与@furas 类似的方法，但略有改进：

    
import os
import gradio as gr
from fastapi import FastAPI, Request
import uvicorn
import threading
from typing import List
from datetime import datetime

api = FastAPI()

# Shared logs

class Log():
    def __init__(self):
        self._logs: List[str] = []
        self.logstr=""
    
    def log_message(self,msg: str):
        timestamp = datetime.now().strftime("%H:%M:%S")
        self._logs.append(f"[{timestamp}] {msg}")
    
    def get_logs(self):
        return "\n".join(self._logs)

log = Log()

def getLog():
    return log.get_logs()

# === FastAPI Setup ===
@api.post("/log")
async def receive_log(request: Request):
    data = await request.body()
    msg = f"API received: {data}"
    log.log_message(msg)
    return {"status": "logged", "message": msg}


def run_api():
    api_port = int(os.environ.get("API_PORT", 8000))
    uvicorn.run(api, host="0.0.0.0", port=api_port)

# === Gradio UI ===

with gr.Blocks() as ui:
    gr.Markdown("## 📝 Incoming HTTP Requests")
    log_box = gr.Textbox(value=getLog, label="Logs", every=1, interactive=False,lines=20)

def run_gradio():
    gradio_port = int(os.environ.get("GRADIO_PORT", 7860))
    ui.queue()
    ui.launch(server_port=gradio_port)

# === Start Both ===
if __name__ == "__main__":
    threading.Thread(target=run_api, daemon=True).start()
    run_gradio()

    我所做的是将日志移到类的字符串生成中Log，并getLog通过调用实际上将日志作为字符串接收log.get_logs。

    • 0

相关问题