FastAPI中如何分块流式传输音频文件？解决返回乱码问题

原代码存在两个核心问题导致异常：

• 未实现真正的流式传输：await f.read()会一次性将整个文件加载到内存，再通过BytesIO返回，本质是一次性返回全部内容，并非分块流式传输。
• 媒体类型不匹配：文件是WAV格式，但返回的media_type设为audio/aac，客户端按AAC编码规则解析WAV二进制数据，导致解析失败出现乱码。

通过异步生成器逐块读取文件内容，配合正确的媒体类型设置，实现真正的流式传输：

完整代码示例 #

from fastapi import StreamingResponse, HTTPException
import aiofiles
import os
# 替换为你的实际模块导入
from your_module import Song, config

async def stream_audio(uid: str):
    track = await Song.find_one({"slug": uid})
    if not track:
        raise HTTPException(status_code=404, detail="音频文件不存在")
    
    # 安全拼接文件路径，避免全局目录变更
    audio_path = os.path.join(config("DLL_AUDIO_FILES"), f"{track.get('file_slug')}.wav")
    
    # 异步生成器：逐块读取文件
    async def audio_generator():
        chunk_size = 1024 * 1024  # 1MB分块，可根据需求调整大小
        async with aiofiles.open(audio_path, mode="rb") as f:
            while chunk := await f.read(chunk_size):
                yield chunk
    
    return StreamingResponse(
        audio_generator(),
        media_type="audio/wav",
        headers={
            "Content-Disposition": f'inline; filename="{track.get("file_slug")}.wav"'
        }
    )

关键改进说明 #

• 异步生成器流式输出：通过audio_generator逐块读取文件，每次返回指定大小的字节块，避免大文件占用过多内存，实现真正的分块流式传输。
• 修正媒体类型：将media_type改为audio/wav（或audio/x-wav），确保客户端能正确识别并解析WAV格式数据。
• 路径安全处理：使用os.path.join拼接文件路径，替代os.chdir的全局目录变更操作，避免路径冲突问题。
• 错误处理增强：增加音频文件不存在的判断，返回404状态码，提升接口健壮性。
• 响应头优化：添加Content-Disposition头，指定文件名并设置inline让客户端直接播放（若需下载可改为attachment）。

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