37. FastAPI 后台任务的使用与完整示例

后台任务在响应之后继续做事

发送通知、写小型审计记录、生成轻量报告，不一定要让用户等待。BackgroundTasks 会在响应发送后执行附加任务，但仍运行在当前应用进程中，不是持久任务队列。

同步任务由线程池执行；异步任务应使用可等待的 I/O。给函数加上 async 后继续调用阻塞库，并不会自动解决阻塞问题。

完整示例：提交导出、查询状态、下载文件

保存为 main.py，默认使用单进程运行。任务状态保存在内存中，输出写入当前目录的 reports。

import csv
import logging
from pathlib import Path
from threading import Lock
from uuid import uuid4
from fastapi import BackgroundTasks, FastAPI, HTTPException
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field

app = FastAPI()
REPORTS = Path("reports")
REPORTS.mkdir(exist_ok=True)
jobs: dict[str, dict] = {}
lock = Lock()
logger = logging.getLogger(__name__)

class ExportIn(BaseModel):
    count: int = Field(default=100, ge=1, le=10000)

def generate_report(job_id: str, count: int):
    temporary = REPORTS / f"{job_id}.tmp"
    target = REPORTS / f"{job_id}.csv"
    try:
        with lock:
            jobs[job_id]["status"] = "running"
        with temporary.open("w", encoding="utf-8-sig", newline="") as file:
            writer = csv.writer(file)
            writer.writerow(["id", "name"])
            for i in range(1, count + 1):
                writer.writerow([i, f"用户{i}"])
        temporary.replace(target)
        with lock:
            jobs[job_id]["status"] = "done"
    except Exception:
        temporary.unlink(missing_ok=True)
        logger.exception("导出失败 job_id=%s", job_id)
        with lock:
            jobs[job_id]["status"] = "failed"

@app.post("/exports", status_code=202)
def create_export(data: ExportIn, background_tasks: BackgroundTasks):
    job_id = uuid4().hex
    with lock:
        jobs[job_id] = {"status": "pending"}
    background_tasks.add_task(generate_report, job_id, data.count)
    return {"job_id": job_id, "status_url": f"/exports/{job_id}"}

@app.get("/exports/{job_id}")
def export_status(job_id: str):
    with lock:
        job = jobs.get(job_id)
        if job is None:
            raise HTTPException(404, "任务不存在")
        return {"job_id": job_id, **job}

@app.get("/exports/{job_id}/file")
def export_file(job_id: str):
    with lock:
        job = jobs.get(job_id)
        if job is None:
            raise HTTPException(404, "任务不存在")
        if job["status"] != "done":
            raise HTTPException(409, "任务尚未成功完成")
    target = REPORTS / f"{job_id}.csv"
    if not target.is_file():
        raise HTTPException(404, "报告文件不存在")
    return FileResponse(target, media_type="text/csv", filename="report.csv")

启动 uv run uvicorn main:app，在 /docs 提交 POST /exports，正文 {"count":100}。拿到任务 ID 后查询状态，完成后下载文件。任务可能很快完成，第一次查询就得到 done 也正常。

为什么返回 202

202 表示请求已接受，工作尚不保证完成。后台失败发生时，原来的响应已经发出，不能再修改成 500，因此要有状态查询和错误记录。示例不将内部异常细节直接公开。

background_tasks.add_task(func, arg) 传函数和参数，不要写成 add_task(func(arg))，那会提前执行。不要把请求的 UploadFile、Request 或数据库 Session 传给长期任务；传 ID、路径等可重新获取资源的数据，任务内部独立打开和关闭资源。

何时改用可靠队列

多 worker 的内存不共享，重启也会丢失 jobs。进程崩溃会丢任务；同一响应附加的任务通常按序执行，一个未处理异常可能影响后续任务。它不适合百万条导入、支付处理、长期模型训练或必须保证送达的邮件。

需要持久化、重试、调度和多进程消费时，使用消息队列与独立 worker，并实现幂等。数据库写入与任务投递要考虑一致性，可评估事务发件箱模式。重试也可能重复执行，不要把队列存在等同于恰好执行一次。

工程扩展与练习

生产导出增加用户身份、任务归属校验、持久任务表、文件有效期与清理；限制同时运行的任务和下载权限，避免内存状态无限增长。锁只保护本进程线程，不解决多实例共享。

练习：故意让输出目录不可写，观察状态转为 failed；重启应用后查询旧任务，理解此示例的持久化边界。使用 TestClient 测试时，调用返回常会等待后台处理完成，不要据此误判真实网络响应时序。