fastapi下载文件接口实现教程

在 FastAPI 中构建文件下载接口，核心方案主要划分为两类：针对服务器本地已存在的实体文件，使用 FileResponse 是最直接高效的途径，框架会自动计算文件长度、推断 MIME 类型并支持断点续传；针对动态生成的大数据报表或海量文件，使用 StreamingResponse 配合异步分块生成器能够以极低的内存开销完成流式回传。结合规范的 Content-Disposition 中文文件名编码（RFC 5987）与 BackgroundTasks 临时文件回收，即可搭建出安全、稳定的生产级下载服务。

from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse

app = FastAPI(title="文件下载服务示例")

# 定义统一的受控存储根目录
STORAGE_DIR = Path("/data/app_storage/files").resolve()

@app.get("/api/v1/download/{filename}")
async def download_file(filename: str):
    # 严格解析目标文件的绝对物理路径
    target_path = (STORAGE_DIR / filename).resolve()
    
    # 路径安全沙箱校验：必须严格归属于 STORAGE_DIR 且为常规实体文件
    if not target_path.is_relative_to(STORAGE_DIR) or not target_path.is_file():
        raise HTTPException(status_code=404, detail="目标文件不存在或无法访问")
        
    return FileResponse(
        path=target_path,
        filename=filename,
        media_type="application/octet-stream"
    )

from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import aiofiles

app = FastAPI()
MEDIA_ROOT = Path("/data/media").resolve()

async def chunk_generator(file_path: Path, chunk_size: int = 65536):
    async with aiofiles.open(file_path, mode="rb") as f:
        while chunk := await f.read(chunk_size):
            yield chunk

@app.get("/api/v1/download-large/{filename}")
async def download_large_file(filename: str):
    target_path = (MEDIA_ROOT / filename).resolve()
    if not target_path.is_relative_to(MEDIA_ROOT) or not target_path.is_file():
        raise HTTPException(status_code=404, detail="文件不存在")

    file_size = target_path.stat().st_size
    headers = {
        "Content-Disposition": f'attachment; filename="{filename}"',
        "Content-Length": str(file_size)
    }

    return StreamingResponse(
        content=chunk_generator(target_path),
        media_type="application/octet-stream",
        headers=headers
    )

技术要点：在 StreamingResponse 中手动添加 Content-Length 标头可以让客户端和浏览器的下载管理器准确显示下载进度百分比和剩余时间。如果省略该标头，响应将进入纯分块编码（Chunked Transfer），浏览器端只能显示已下载字节数，无法预估总体进度。

import io
import csv
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

def generate_csv_stream():
    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(["用户ID", "姓名", "注册时间", "账户状态"])
    yield output.getvalue().encode("utf-8-sig")
    output.seek(0)
    output.truncate(0)

    # 模拟从数据库分页查询并持续推送
    for batch_id in range(100):
        records = [
            [f"U{batch_id * 10 + i}", f"用户_{i}", "2026-09-30", "正常"]
            for i in range(10)
        ]
        for row in records:
            writer.writerow(row)
        yield output.getvalue().encode("utf-8-sig")
        output.seek(0)
        output.truncate(0)

@app.get("/api/v1/export/users.csv")
async def export_users():
    headers = {
        "Content-Disposition": 'attachment; filename="users.csv"'
    }
    return StreamingResponse(
        content=generate_csv_stream(),
        media_type="text/csv",
        headers=headers
    )

from urllib.parse import quote
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

def build_content_disposition(chinese_filename: str) -> str:
    # 对中文部分进行标准 URL 编码
    encoded_filename = quote(chinese_filename)
    # 构造同时满足基础标准与 RFC 5987 国际化标准的复合标头
    return f"attachment; filename=\"download\"; filename*=utf-8''{encoded_filename}"

@app.get("/api/v1/download-chinese")
async def download_chinese_file():
    real_name = "2026年第三季度财务审计决算报表.xlsx"
    headers = {
        "Content-Disposition": build_content_disposition(real_name)
    }
    # 演示返回空流
    return StreamingResponse(
        content=iter([b"mock data"]),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers=headers
    )

import os
from pathlib import Path
from fastapi import FastAPI, BackgroundTasks, HTTPException
from fastapi.responses import FileResponse

app = FastAPI()

def cleanup_file(file_path: Path):
    # 安全回收临时文件的后台回调任务
    try:
        if file_path.exists() and file_path.is_file():
            file_path.unlink()
    except OSError:
        pass  # 记录日志，避免后台清理异常影响主流程

@app.get("/api/v1/download-archive")
async def download_archive(background_tasks: BackgroundTasks):
    temp_zip = Path("/tmp/export_session_957c.zip")
    
    # 模拟生成归档压缩包
    temp_zip.write_bytes(b"PK...archive mock payload...")
    
    # 注册响应发送完毕后的清理任务
    background_tasks.add_task(cleanup_file, temp_zip)
    
    return FileResponse(
        path=temp_zip,
        filename="系统运行归档.zip",
        media_type="application/zip",
        background=background_tasks
    )