Início / coding / Perguntas / 79598793
Accepted
Dimitrios Desyllas
Asked: 2025-04-30 00:08:50 +0800 CST

Como posso atualizar a interface do usuário em tempo real quando recebo uma solicitação no FastAPI?

  • 772

Eu tenho este script simples:

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()

O que tento fazer é ter o FastAPI escutando uma porta e um painel que exibe em tempo real as solicitações recebidas:

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

Mas não consigo alterar o conteúdo Textboxquando recebo solicitações de entrada no FastAPI. Como posso fazer isso?

python

2 respostas

  1. Best Answer

    O único método que funciona para mim é

    def function():
    return log.logstr

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

    Ele é executado functiona cada 1 segundo e functionretorna o conteúdo atual no log.

    Doc: Caixa de texto Gradio

    Código completo:

    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. Uma abordagem semelhante com pequenas melhorias foi seguida por @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()

    O que fiz foi mover o log para a geração de strings na Logclasse e, getLogna verdade, receber o log como uma string chamando log.get_logs.

    • 0

relate perguntas