如何在FastAPI中实现Excel文件流式下载以触发浏览器立即下载

问题描述 #

我正在构建FastAPI应用，需要实现一个允许用户通过流式传输下载Excel文件的端点，目标是用户点击浏览器中的下载按钮后，文件立即开始下载，如同常规网站的文件下载体验。

当前实现代码 #

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import openpyxl
import io
import time

app = FastAPI()

@app.get("/download_excel/")
async def get_attendance_report():
    start_time = time.time()
    
    # Create Excel file in memory
    wb = openpyxl.Workbook()
    ws = wb.active

    headers = ["id", "name", "age", "city", "is_student"]
    ws.append(headers)

    # Sample data
    data = [
        {'id': 1, 'name': 'Alice', 'age': 30, 'city': 'New York', 'is_student': True},
        {'id': 2, 'name': 'Bob', 'age': 25, 'city': 'Los Angeles', 'is_student': False}
    ]
    
    for record in data:
        ws.append([record['id'], record['name'], record['age'], record['city'], record['is_student']])

    # log.info(f"File generation time: {time.time() - start_time} seconds")

    def iterfile():
        with io.BytesIO() as byte_stream:
            wb.save(byte_stream)
            byte_stream.seek(0)
            while chunk := byte_stream.read(262144):  # 256KB chunks
                yield chunk

    headers = {
        'Content-Disposition': 'attachment; filename="employee_attendance_report.xlsx"',
        'Transfer-Encoding': 'chunked'
    }
    
    return StreamingResponse(iterfile(), media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers=headers)

遇到的问题 #

• 点击下载按钮后，立即收到状态码200，但下载进程启动需要较长时间。
• 希望点击按钮后文件立即开始下载，在浏览器下载管理器中显示文件名及下载进度。

已尝试方案 #

• 采用StreamingResponse实现文件分块流式传输
• 设置Content-Disposition头指定附件及文件名

疑问 #

1. 针对Excel文件的StreamingResponse实现是否正确？
2. 是否需要额外的HTTP头或配置以确保浏览器立即启动下载？
3. 如何缩短收到状态码与下载启动之间的延迟？

解决方案 #

1. 现有实现的核心问题 #

当前代码的StreamingResponse用法未发挥流式传输的本质：Excel文件的创建、数据写入、内存保存全在返回响应前完成，相当于服务器先把整个文件生成好，再分块发送。这就导致用户点击下载后，必须等待文件完全生成才能开始接收数据，自然出现延迟。

2. 正确的流式下载实现 #

要让浏览器立即启动下载，需将Excel生成逻辑移到StreamingResponse的迭代器函数内，让服务器先返回响应头（包含文件名等信息），再边生成文件边发送数据。

修改后的openpyxl版本代码

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

app = FastAPI()

@app.get("/download_excel/")
async def get_attendance_report():
    def generate_excel_stream():
        # 在迭代器内完成Excel生成，避免提前占用内存和时间
        wb = openpyxl.Workbook()
        ws = wb.active
        
        # 写入表头
        headers = ["id", "name", "age", "city", "is_student"]
        ws.append(headers)
        
        # 写入数据（若数据来自数据库，可改为边查询边写入）
        data = [
            {'id': 1, 'name': 'Alice', 'age': 30, 'city': 'New York', 'is_student': True},
            {'id': 2, 'name': 'Bob', 'age': 25, 'city': 'Los Angeles', 'is_student': False}
        ]
        
        for record in data:
            ws.append([record['id'], record['name'], record['age'], record['city'], record['is_student']])
        
        # 保存到内存流并分块发送
        with io.BytesIO() as byte_stream:
            wb.save(byte_stream)
            byte_stream.seek(0)
            while chunk := byte_stream.read(262144):
                yield chunk

    headers = {
        'Content-Disposition': 'attachment; filename="employee_attendance_report.xlsx"',
        'Cache-Control': 'no-cache'  # 避免浏览器缓存旧文件
    }
    
    return StreamingResponse(
        generate_excel_stream(),
        media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        headers=headers
    )

更高效的xlsxwriter版本（支持内存流式写入）

如果数据量较大，推荐使用xlsxwriter库，它支持直接在内存中生成Excel并流式输出，进一步降低延迟：

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

app = FastAPI()

@app.get("/download_excel/")
async def get_attendance_report():
    def generate_excel_stream():
        output = io.BytesIO()
        # 启用内存模式生成Excel
        workbook = xlsxwriter.Workbook(output, {'in_memory': True})
        worksheet = workbook.add_worksheet()
        
        # 写入表头
        headers = ["id", "name", "age", "city", "is_student"]
        for col_num, header in enumerate(headers):
            worksheet.write(0, col_num, header)
        
        # 写入数据
        data = [
            {'id': 1, 'name': 'Alice', 'age': 30, 'city': 'New York', 'is_student': True},
            {'id': 2, 'name': 'Bob', 'age': 25, 'city': 'Los Angeles', 'is_student': False}
        ]
        
        for row_num, record in enumerate(data, start=1):
            worksheet.write(row_num, 0, record['id'])
            worksheet.write(row_num, 1, record['name'])
            worksheet.write(row_num, 2, record['age'])
            worksheet.write(row_num, 3, record['city'])
            worksheet.write(row_num, 4, record['is_student'])
        
        workbook.close()
        output.seek(0)
        
        # 分块发送数据
        while chunk := output.read(262144):
            yield chunk

    headers = {
        'Content-Disposition': 'attachment; filename="employee_attendance_report.xlsx"',
        'Cache-Control': 'no-cache'
    }
    
    return StreamingResponse(
        generate_excel_stream(),
        media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        headers=headers
    )

3. 关键优化说明 #

• 移除手动设置的Transfer-Encoding: chunked：FastAPI的StreamingResponse会自动处理分块传输，手动设置可能导致浏览器解析异常。
• 迭代器内生成文件：确保服务器先返回响应头，浏览器收到后立刻显示下载进度和文件名，再逐步接收文件数据。
• 流式数据处理：如果数据来自数据库，使用流式查询（如SQLAlchemy的yield_per），边查询边写入Excel，避免一次性加载大量数据到内存。

4. 额外配置建议 #

• 添加Cache-Control: no-cache头，确保每次下载都获取最新文件，避免浏览器缓存旧数据。
• 若文件大小可预估，添加Content-Length头，让浏览器更准确显示下载进度（但流式生成时无法提前获取，此条仅适用于已知文件大小的场景）。

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