Streamlit+FastAPI如何处理10分钟级长运行任务并返回结果？

关于BackgroundTasks的状态追踪 #

FastAPI原生的BackgroundTasks是轻量级临时任务实现，没有内置的任务ID和状态追踪能力。它的任务绑定在请求上下文里，请求返回后任务就脱离可追踪范围，无法主动查询状态、进度或结果。要实现长任务的状态追踪，必须替换为更专业的任务管理方案。

更优实现方案 #

你的初始思路可行，但可以从后端任务管理、前端交互体验两方面优化：

后端优化：用任务队列或数据库追踪状态 #

方案1：使用专业任务队列（推荐）

针对10分钟的长任务，最稳妥的方案是采用Celery + Redis/RabbitMQ作为消息中间件。每个任务会自动生成唯一task_id，支持：

• 通过task_id直接查询任务状态（PENDING、SUCCESS、FAILURE、PROGRESS等）
• 内置重试、超时、任务优先级管理
• 支持分布式部署（方便后续扩容）

核心流程：

• 上传文件接口：接收Excel后，将文件路径和处理逻辑封装为Celery任务，启动任务并返回task_id
• 任务状态/结果接口：通过task_id调用Celery的AsyncResult获取状态，任务完成后返回处理好的DataFrame

方案2：轻量方案——数据库存任务状态

如果不想引入Celery这类重型组件，可以用数据库（SQLite/PostgreSQL）维护任务生命周期：

1. 上传文件时用uuid.uuid4()生成唯一task_id
2. 在数据库中插入任务记录：包含task_id、状态（初始为“pending”）、文件存储路径、结果字段（为空）
3. 启动后台任务（可用BackgroundTasks或threading.Thread），处理过程中更新数据库状态为“processing”，完成后将DataFrame序列化（如转JSON或存为Parquet文件路径）并更新状态为“completed”
4. 提供GET /tasks/{task_id}接口，返回当前状态；若任务完成，则附带处理结果

前端优化：提升交互效率 #

1. 替换固定轮询为指数退避轮询：不要每分钟固定查询，而是先以短间隔（如10秒）查询，之后逐渐延长至1分钟，减少不必要的服务器请求。例如第一次等10秒，第二次20秒，第三次40秒，之后保持60秒间隔。
2. 动态状态提示：在Streamlit中用st.empty()创建占位容器，每次轮询后更新状态文本（如“处理中...已完成50%”，若后端能返回进度的话），让用户有明确感知。
3. 可选：WebSocket实时推送：若想彻底避免轮询，可在后端实现WebSocket端点，任务处理时主动向前端推送状态更新。Streamlit可通过自定义组件或嵌入JS实现WebSocket监听，实时更新页面状态。

代码示例（轻量方案） #

后端（FastAPI + SQLite） #

from fastapi import FastAPI, BackgroundTasks, UploadFile, File
from sqlalchemy import create_engine, Column, String, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import uuid
import pandas as pd
import time

# 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 任务模型
class Task(Base):
    __tablename__ = "tasks"
    task_id = Column(String, primary_key=True, index=True)
    status = Column(String, default="pending")
    result = Column(Text, nullable=True)

Base.metadata.create_all(bind=engine)

app = FastAPI()

def process_task(task_id: str, file_path: str):
    db = SessionLocal()
    try:
        # 更新状态为处理中
        db.query(Task).filter(Task.task_id == task_id).update({"status": "processing"})
        db.commit()
        
        # 模拟10分钟处理（实际替换为你的业务逻辑）
        df = pd.read_excel(file_path)
        # 此处编写数据清洗、计算等业务逻辑
        time.sleep(600)
        
        # 序列化结果（大数据可改为存储文件路径）
        result_json = df.to_json(orient="records")
        # 更新状态和结果
        db.query(Task).filter(Task.task_id == task_id).update({"status": "completed", "result": result_json})
        db.commit()
    except Exception as e:
        db.query(Task).filter(Task.task_id == task_id).update({"status": f"failed: {str(e)}"})
        db.commit()
    finally:
        db.close()

@app.post("/upload")
async def upload_file(file: UploadFile = File(...), background_tasks: BackgroundTasks = None):
    # 保存上传文件
    file_path = f"./uploads/{file.filename}"
    with open(file_path, "wb") as f:
        f.write(await file.read())
    
    # 生成任务ID并初始化记录
    task_id = str(uuid.uuid4())
    db = SessionLocal()
    db.add(Task(task_id=task_id))
    db.commit()
    db.close()
    
    # 启动后台任务
    background_tasks.add_task(process_task, task_id, file_path)
    return {"task_id": task_id}

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    db = SessionLocal()
    task = db.query(Task).filter(Task.task_id == task_id).first()
    db.close()
    if not task:
        return {"status": "not_found"}
    response = {"task_id": task_id, "status": task.status}
    if task.status == "completed":
        response["result"] = pd.read_json(task.result).to_dict(orient="records")
    return response

前端（Streamlit） #

import streamlit as st
import requests
import time
import pandas as pd

st.title("Excel文件处理工具")

# 文件上传组件
uploaded_file = st.file_uploader("选择Excel文件", type=["xlsx", "xls"])

if uploaded_file is not None:
    # 提交文件到后端
    if st.button("开始处理"):
        files = {"file": (uploaded_file.name, uploaded_file, uploaded_file.type)}
        response = requests.post("http://localhost:8000/upload", files=files)
        task_id = response.json()["task_id"]
        st.session_state["task_id"] = task_id
        st.session_state["status"] = "pending"
    
    # 轮询任务状态
    if "task_id" in st.session_state:
        status_placeholder = st.empty()
        backoff = 10  # 初始等待10秒
        max_backoff = 60  # 最大等待60秒
        
        while st.session_state["status"] not in ["completed", "failed", "not_found"]:
            response = requests.get(f"http://localhost:8000/tasks/{st.session_state['task_id']}")
            data = response.json()
            st.session_state["status"] = data["status"]
            
            status_placeholder.text(f"当前状态: {data['status']}")
            
            if st.session_state["status"] == "processing":
                time.sleep(backoff)
                backoff = min(backoff * 2, max_backoff)
            else:
                break
        
        # 处理结果展示
        if st.session_state["status"] == "completed":
            st.success("处理完成！")
            df = pd.DataFrame(data["result"])
            st.dataframe(df)
            # 示例图表绘制
            st.bar_chart(df)
        elif st.session_state["status"].startswith("failed"):
            st.error(f"处理失败: {st.session_state['status'].split(':')[1]}")

内容的提问来源于stack exchange，提问作者Nicolas