如何将作为 zip 文件的 FastAPI UploadFile 以 .zip 格式保存到磁盘？

Question

我正在通过 FastAPI 将 zip 文件上传为 UploadFile，并希望使用 async aiofiles 将它们保存到文件系统中，如下所示：

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)
        await f.close()

文件创建于 filepath，但它的大小为 0B，unzip out_file.zip 产生以下错误：

Archive: out_file.zip
    End-of-central-directory signature not found. Either this file is not
    a zipfile, or it constitutes one disk of a multi-part archive. In the
    latter case the central directory and zipfile comment will be found on
    the last disk(s) of this archive.
unzip:  cannot find zipfile directory in one of out_file.zip or out_file.zip.zip,
        and cannot find out_file.zip.ZIP, period.

print(in_file.content_type) 输出 application/x-zip-compressed 和

python3 -m mimetypes out_file.zip 产生 类型：application/zip 编码：None

我在这种不便上花费了太多时间，并尝试了几种阻止替代方法，例如：

with open(filepath, "wb") as f:
    f.write(in_file.file.read())
    f.close()

结果都是一样的。我现在正在尝试使用 .zip 文件来实现这一点，但最终我正在寻找二进制文件的通用解决方案以在它们出现时保存它们，因为我没有处理任何文件，它们只需要存储。

如果有人能指出我遗漏了什么，那将会很有帮助。

编辑： 在我尝试将文件写入我的文件系统之前，我通过 Motor:

添加了一个带有一些元数据的条目到我的数据库中

@router.post("/")
async def upload(in_file: UploadFile = File(...)):
    file_content = await in_file.read()
    file_db = {"name": in_file.filename, "size": len(file_content)}
    file_db_json = jsonable_encoder(file_db)
    added_file_db = await add_file(file_db_json) 

    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)
        
    return ResponseModel(added_file_db, "upload successful")

upload() 中的 return 确认上传成功，元数据已添加到数据库，文件已在我的文件系统中创建但已损坏，如上所述。我不知道这会如何干扰将文件内容写入我的磁盘，但也许我错了。

Answer 1

使用如下（取自this answer）：

import aiofiles
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    async with aiofiles.open(file.filename, 'wb') as f:
        while content := await file.read(1024): # async read chunk
            await f.write(content)
        
    return {"Uploaded File": file.filename}

如果您需要将文件保存在特定目录中，请使用以下：

import aiofiles
import os
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    filename = os.path.join('path/to/', file.filename) 
    async with aiofiles.open(filename, 'wb') as f:
        while content := await file.read(1024): # async read chunk
            await f.write(content)
        
    return {"Uploaded File": file.filename}

更新

您问题中最近的编辑表明您已经读取了第 file_content = await in_file.read() 行的文件内容，因此，尝试使用 await in_file.read(1024) 再次读取内容会导致读取零字节。因此，要么在读取和保存文件后将元数据添加到数据库（您可以使用变量来保持总文件长度，例如 total_len += len(buffer)），或者只将 file_content 写入本地文件，如下图

async def upload(file: UploadFile = File(...)):
    filename = os.path.join('path/to/', file.filename) 
    async with aiofiles.open(filename, 'wb') as f:
        await f.write(file_content)
        
    return {"Uploaded Filename": file.filename}

更新 2

为了完整起见，我还应该提到有一个内部“游标”（或“文件指针”）表示文件内容将被读取（或写入）的位置。当调用 read() 时，一直读取到缓冲区的末尾，在光标之外留下零字节。因此，也可以使用 seek() 方法将光标的当前位置设置为 0（即，将光标倒回到文件的开头）。根据 FastAPI documentation:

seek(offset): Goes to the byte position offset (int) in the file.

E.g., await myfile.seek(0) would go to the start of the file.

This is especially useful if you run await myfile.read() once and then need to read the contents again.

如何将作为 zip 文件的 FastAPI UploadFile 以 .zip 格式保存到磁盘？

How do I save a FastAPI UploadFile which is a zip file to disk as .zip?

python

filesystems

zip

python-3.x

fastapi

更新

更新 2