FastAPI背景任务完成后如何跳转至指定HTML页面？

FastAPI的BackgroundTask是在服务器返回响应之后才开始执行的，而HTTP是无状态协议——一旦服务器把提示“后台运行”的HTMLResponse发给浏览器，两者的连接就断开了，服务器没办法主动触发浏览器的跳转操作。要实现任务完成后的跳转，得通过前端主动检测或服务器主动推送的方式来实现，下面是两种可行的方案：

方案1：前端轮询检查任务状态 #

这是最易实现的方案，核心思路是让前端定时向服务器查询任务进度，检测到完成后自行跳转。

服务器端代码 #

from fastapi import FastAPI, BackgroundTasks, Form
from fastapi.responses import HTMLResponse
import time
import uuid

app = FastAPI()

# 开发环境用字典存任务状态，生产环境建议改用Redis/数据库
task_status = {}

def background_task(task_id: str):
    # 模拟耗时业务逻辑
    time.sleep(5)
    task_status[task_id] = "completed"

# 初始表单页面
@app.get("/", response_class=HTMLResponse)
async def index():
    return """
    <form action="/submit" method="post">
        <input type="text" name="info" placeholder="输入信息">
        <button type="submit">提交</button>
    </form>
    """

# 表单提交接口，返回等待页面+轮询逻辑
@app.post("/submit", response_class=HTMLResponse)
async def submit(info: str = Form(...), background_tasks: BackgroundTasks = BackgroundTasks()):
    task_id = str(uuid.uuid4())
    task_status[task_id] = "running"
    background_tasks.add_task(background_task, task_id)
    
    return f"""
    <html>
        <body>
            <p>程序正在后台运行，请稍候...</p>
            <script>
                // 每1秒查询一次任务状态
                setInterval(async () => {{
                    const res = await fetch(`/check-task/{task_id}`);
                    const status = await res.text();
                    if (status === 'completed') {{
                        window.location.href = '/success-page';
                    }}
                }}, 1000);
            </script>
        </body>
    </html>
    """

# 任务状态查询接口
@app.get("/check-task/{task_id}")
async def check_task(task_id: str):
    return task_status.get(task_id, "running")

# 任务完成后的跳转页面
@app.get("/success-page", response_class=HTMLResponse)
async def success_page():
    return "<html><body><h1>任务完成！</h1></body></html>"

方案2：使用Server-Sent Events(SSE)实现主动推送 #

如果想减少前端的无效请求，可以用SSE让服务器在任务完成时主动通知前端，适合对实时性要求稍高的场景。

服务器端代码 #

from fastapi import FastAPI, BackgroundTasks, Form
from fastapi.responses import HTMLResponse
from sse_starlette.sse import EventSourceResponse
import time
import uuid
import asyncio

app = FastAPI()

task_status = {}

def background_task(task_id: str):
    time.sleep(5)
    task_status[task_id] = "completed"

@app.get("/", response_class=HTMLResponse)
async def index():
    return """
    <form action="/submit-sse" method="post">
        <input type="text" name="info" placeholder="输入信息">
        <button type="submit">提交</button>
    </form>
    """

@app.post("/submit-sse", response_class=HTMLResponse)
async def submit_sse(info: str = Form(...), background_tasks: BackgroundTasks = BackgroundTasks()):
    task_id = str(uuid.uuid4())
    task_status[task_id] = "running"
    background_tasks.add_task(background_task, task_id)
    
    return f"""
    <html>
        <body>
            <p>程序正在后台运行，请稍候...</p>
            <script>
                const eventSource = new EventSource(`/task-status/{task_id}`);
                eventSource.onmessage = function(event) {{
                    if (event.data === 'completed') {{
                        eventSource.close();
                        window.location.href = '/success-page';
                    }}
                }};
            </script>
        </body>
    </html>
    """

@app.get("/task-status/{task_id}")
async def task_status_sse(task_id: str):
    async def event_generator():
        while True:
            status = task_status.get(task_id, "running")
            yield {"data": status}
            if status == "completed":
                break
            await asyncio.sleep(1)
    return EventSourceResponse(event_generator())

@app.get("/success-page", response_class=HTMLResponse)
async def success_page():
    return "<html><body><h1>任务完成！</h1></body></html>"

生产环境注意事项 #

• 不要用内存字典task_status存储任务状态，建议改用Redis、MongoDB等持久化存储方案
• 轮询间隔可根据任务实际耗时调整，避免过于频繁的请求浪费资源
• SSE需要浏览器支持，现代主流浏览器均已兼容，但如果要兼容老旧浏览器，优先选择轮询方案

内容的提问来源于stack exchange，提问作者suspicious-nebula