使用 FastAPI 构建轻量级文件处理服务：从上传到任务下载

一、前言

在开发 PDF 工具、图片处理工具或文档转换应用时，前端通常只负责选择文件和展示结果，真正的文件处理需要放在后端完成。

本文介绍如何使用 Python FastAPI 构建一个轻量级文件处理服务，实现以下流程：

上传文件 → 创建任务 → 后台处理 → 查询状态 → 下载结果

这套方式适合个人项目、小程序后端和内部工具。

二、为什么选择 FastAPI

FastAPI 具有以下特点：

• 开发速度快；
• 自带 Swagger 接口文档；
• 支持异步接口；
• 参数校验方便；
• 适合构建文件上传和 API 服务。

安装依赖：

pip install fastapi uvicorn python-multipart

如果需要处理 PDF 和图片，可以继续安装：

pip install Pillow PyMuPDF

三、创建基础项目

项目结构如下：

pdf-service/
├── app.py
├── data/
└── requirements.txt

创建 app.py：

from pathlib import Path
from fastapi import FastAPI, UploadFile, File, HTTPException
import uuid

app = FastAPI(title="Document Service")

DATA_DIR = Path("data")
DATA_DIR.mkdir(exist_ok=True)

tasks = {}

这里使用 data 目录保存上传文件和处理结果。实际生产环境中，可以替换为对象存储。

四、实现文件上传

添加上传接口：

@app.post("/api/tasks")
async def create_task(
    file: UploadFile = File(...)
):
    if not file.filename:
        raise HTTPException(status_code=400, detail="文件名不能为空")

    task_id = uuid.uuid4().hex
    task_dir = DATA_DIR / task_id
    task_dir.mkdir(parents=True, exist_ok=True)

    file_path = task_dir / Path(file.filename).name

    with file_path.open("wb") as output:
        while chunk := await file.read(1024 * 1024):
            output.write(chunk)

    tasks[task_id] = {
        "id": task_id,
        "status": "queued",
        "filename": file.filename
    }

    return {
        "task_id": task_id,
        "status": "queued"
    }

这里每个任务都有独立目录，能够避免不同用户上传的文件互相覆盖。

同时使用：

Path(file.filename).name

可以避免用户通过文件名传入路径，降低路径穿越风险。

五、实现任务处理

下面以计算文件 MD5 为例：

import hashlib

def calculate_md5(file_path: Path) -> str:
    md5 = hashlib.md5()

    with file_path.open("rb") as file:
        while chunk := file.read(1024 * 1024):
            md5.update(chunk)

    return md5.hexdigest()

创建处理函数：

def process_task(task_id: str, file_path: Path):
    try:
        tasks[task_id]["status"] = "processing"

        digest = calculate_md5(file_path)
        result_path = file_path.parent / "result.txt"
        result_path.write_text(digest, encoding="utf-8")

        tasks[task_id]["status"] = "success"
        tasks[task_id]["result"] = str(result_path)

    except Exception as error:
        tasks[task_id]["status"] = "failed"
        tasks[task_id]["error"] = str(error)

实际项目中，可以将 calculate_md5 替换为：

• PDF 合并；
• PDF 拆分；
• 图片压缩；
• 图片转 PDF；
• 添加水印；
• PDF 文本提取。

六、使用后台任务

文件处理可能需要几秒甚至几分钟，不建议让上传接口一直等待。

FastAPI 可以使用后台任务：

from fastapi import BackgroundTasks

@app.post("/api/tasks")
async def create_task(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...)
):
    if not file.filename:
        raise HTTPException(status_code=400, detail="文件名不能为空")

    task_id = uuid.uuid4().hex
    task_dir = DATA_DIR / task_id
    task_dir.mkdir(parents=True, exist_ok=True)

    file_path = task_dir / Path(file.filename).name

    with file_path.open("wb") as output:
        while chunk := await file.read(1024 * 1024):
            output.write(chunk)

    tasks[task_id] = {
        "id": task_id,
        "status": "queued",
        "filename": file.filename
    }

    background_tasks.add_task(process_task, task_id, file_path)

    return {
        "task_id": task_id,
        "status": "queued"
    }

这样接口会快速返回任务编号，客户端可以通过任务编号查询处理进度。

七、查询任务状态

@app.get("/api/tasks/{task_id}")
def get_task(task_id: str):
    task = tasks.get(task_id)

    if not task:
        raise HTTPException(status_code=404, detail="任务不存在")

    return task

可能返回：

{
  "id": "abc123",
  "status": "processing",
  "filename": "example.pdf"
}

处理完成后：

{
  "id": "abc123",
  "status": "success",
  "filename": "example.pdf",
  "result": "data/abc123/result.txt"
}

八、实现结果下载

from fastapi.responses import FileResponse

@app.get("/api/tasks/{task_id}/download")
def download_result(task_id: str):
    task = tasks.get(task_id)

    if not task or task.get("status") != "success":
        raise HTTPException(status_code=404, detail="结果尚未生成")

    result_path = Path(task["result"])

    if not result_path.exists():
        raise HTTPException(status_code=404, detail="结果文件不存在")

    return FileResponse(
        result_path,
        filename=result_path.name
    )

前端拿到任务编号后，可以按照以下顺序执行：

1. 上传文件
2. 保存 task_id
3. 每隔一秒查询任务状态
4. status 为 success 时下载结果
5. status 为 failed 时提示错误

九、生产环境需要注意的问题

1. 限制文件大小

MAX_FILE_SIZE = 20 * 1024 * 1024

不要允许用户无限上传文件，否则很容易占满磁盘。

2. 检查文件类型

不能只依赖文件扩展名，还应该检查文件实际格式，避免恶意文件伪装。

3. 临时文件自动清理

处理完成后的临时文件不应永久保存，可以设置 24 小时自动删除。

4. 不要把任务状态只放在内存中

示例使用内存字典，服务器重启后数据会丢失。正式环境建议使用：

• SQLite；
• MySQL；
• Redis；
• PostgreSQL。

5. 使用 HTTPS

小程序正式调用接口时，必须使用 HTTPS 域名，不能长期使用公网 IP 加 HTTP 端口。

6. 任务数量需要限制

对于配置较低的服务器，可以限制同时处理的任务数量，避免 CPU 和内存被占满。

十、启动服务

开发环境启动命令：

uvicorn app:app --host 0.0.0.0 --port 8000

启动后访问：

http://服务器IP:8000/docs

FastAPI 会自动生成接口测试页面，可以直接上传文件并调用接口。

十一、总结

使用 FastAPI 构建文件处理服务时，建议将上传、处理、查询和下载拆分成不同阶段：

上传接口负责接收文件
任务接口负责记录状态
后台程序负责处理文件
下载接口负责返回结果

这种结构简单清晰，适合小程序、个人工具和轻量级办公应用。

当用户数量增加后，可以进一步加入：

• Redis 任务队列；
• Celery Worker；
• 对象存储；
• 用户登录；
• 文件权限控制；
• 任务失败重试；
• 日志和监控系统。