如何在FastAPI中通过StreamingResponse从Pandas Dataframe返回CSV？

错误原因 #

你遇到的TypeError是因为直接将Dataframe或原始数据字典传给了StreamingResponse，但它要求接收的是可迭代的字符串/字节流对象，而非原始数据结构。

核心解决方案 #

无需写入磁盘，利用Pandas的to_csv()将Dataframe输出到内存缓冲区（io.StringIO或io.BytesIO），再将缓冲区作为流返回给客户端。同时适配异步获取数据的场景。

完整代码示例 #

1. 异步数据获取函数 #

import pandas as pd
import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import io

async def fetch_data():
    # 替换为你的实际异步数据获取逻辑
    await asyncio.sleep(1)
    sample_data = {
        "name": ["Alice", "Bob", "Charlie"],
        "age": [25, 30, 35],
        "city": ["New York", "London", "Paris"]
    }
    return pd.DataFrame(sample_data)

2. CSV流返回接口 #

app = FastAPI()

@app.get("/export-csv")
async def export_csv():
    # 异步获取Dataframe
    df = await fetch_data()
    
    # 初始化内存缓冲区并写入CSV内容
    csv_buffer = io.StringIO()
    df.to_csv(csv_buffer, index=False)
    # 将缓冲区指针移至起始位置，确保能读取全部内容
    csv_buffer.seek(0)
    
    # 返回流式响应，设置正确的媒体类型和下载文件名
    return StreamingResponse(
        csv_buffer,
        media_type="text/csv",
        headers={"Content-Disposition": 'attachment; filename="exported_data.csv"'}
    )

关键细节提醒 #

• 若需处理非ASCII字符，改用io.BytesIO并指定编码：

csv_buffer = io.BytesIO()
df.to_csv(csv_buffer, index=False, encoding="utf-8")
csv_buffer.seek(0)

• 必须调用seek(0)，否则缓冲区读取会从末尾开始，返回空内容。
• 避免直接传递Dataframe或原始数据字典给StreamingResponse，这是触发错误的核心原因。

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