FastAPI中使用aiofiles保存UploadFile类型Zip文件至磁盘失败的问题排查求助

哦，我一眼就看穿问题所在了——你在保存文件之前已经把UploadFile的数据流读空了！

问题根源 #

当你调用await in_file.read()时，FastAPI底层的UploadFile（基于Starlette实现）会把整个文件内容加载到内存里，同时将文件指针移动到了数据流的末尾。这时候你再去循环执行await in_file.read(1024)，只能读到空字节，自然写出来的文件是0B，解压时肯定会报"找不到中央目录签名"的错误——因为文件根本没有实际内容。

几种可行的解决方案 #

根据你的需求（通用二进制文件存储，可选先处理元数据），这里提供3种方案：

方案1：先保存文件，再计算元数据（推荐，适合大文件）

如果上传的文件体积较大，把整个文件读进内存可能导致OOM，所以优先选择先分块保存到磁盘，再从磁盘获取文件大小：

import os
from fastapi import File, UploadFile
from fastapi.responses import ResponseModel
import aiofiles
from your_module import add_file, jsonable_encoder

@router.post("/")
async def upload(in_file: UploadFile = File(...)):
    filepath = "/path/to/out_file.zip"
    
    # 第一步：异步分块保存文件到磁盘
    async with aiofiles.open(filepath, 'wb') as f:
        while buffer := await in_file.read(1024):
            await f.write(buffer)
    
    # 第二步：从磁盘获取文件大小，避免内存占用
    file_size = os.path.getsize(filepath)
    file_db = {"name": in_file.filename, "size": file_size}
    
    # 处理数据库逻辑
    file_db_json = jsonable_encoder(file_db)
    added_file_db = await add_file(file_db_json)
    
    return ResponseModel(added_file_db, "upload successful")

方案2：读取内容后重置文件指针

如果你必须先读取文件内容（比如要做内容校验），可以在读取后手动把文件指针重置到开头，这样后续还能正常读取数据流：

from fastapi import File, UploadFile
from fastapi.responses import ResponseModel
import aiofiles
from your_module import add_file, jsonable_encoder

@router.post("/")
async def upload(in_file: UploadFile = File(...)):
    # 读取全部内容到内存
    file_content = await in_file.read()
    file_size = len(file_content)
    
    # 关键操作：重置文件指针到数据流开头
    await in_file.seek(0)
    
    # 现在可以正常保存文件了
    filepath = "/path/to/out_file.zip"
    async with aiofiles.open(filepath, 'wb') as f:
        while buffer := await in_file.read(1024):
            await f.write(buffer)
    
    # 处理数据库逻辑
    file_db = {"name": in_file.filename, "size": file_size}
    file_db_json = jsonable_encoder(file_db)
    added_file_db = await add_file(file_db_json)
    
    return ResponseModel(added_file_db, "upload successful")

方案3：直接用已读取的内容写入（简单但不适合大文件）

既然已经把文件内容读到file_content变量里了，直接写入磁盘就行，不用二次读取数据流：

from fastapi import File, UploadFile
from fastapi.responses import ResponseModel
import aiofiles
from your_module import add_file, jsonable_encoder

@router.post("/")
async def upload(in_file: UploadFile = File(...)):
    file_content = await in_file.read()
    file_size = len(file_content)
    
    filepath = "/path/to/out_file.zip"
    async with aiofiles.open(filepath, 'wb') as f:
        await f.write(file_content)
    
    # 处理数据库逻辑
    file_db = {"name": in_file.filename, "size": file_size}
    file_db_json = jsonable_encoder(file_db)
    added_file_db = await add_file(file_db_json)
    
    return ResponseModel(added_file_db, "upload successful")

额外注意点 #

• 对于大文件，优先选方案1，方案2和3会把整个文件加载到内存，容易引发内存不足问题；
• 你之前尝试的同步写法失败，原因也是一样的：in_file.file.read()读完后文件指针在末尾，需要调用in_file.file.seek(0)重置指针才能重新读取；
• 这个方案是通用的，不管上传的是ZIP、图片还是其他二进制文件，都能正常保存原始内容。

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