从作为 zip 的 SpooledTemporaryFile 获取信息

Getting information from SpooledTemporaryFile that is a zip

我正在使用 Fast API 上传一个 zip 文件,它将文件作为 SpooledTemporaryFile 接收。我一直在尝试将该文件复制到磁盘上。我尝试过各种各样的事情;我觉得我取得了进步的两个在下面。

  1. 我尝试使用 zipfile.Zipfile,当我把它变成一个时,我无法解压它,因为它说它不是一个 zip。

  2. 我还上传了一个未压缩的文件并将 SpooledTemporaryFile 转换为 _io.BytesIO,但是,当我尝试读取单个文件(未压缩)的内容时,它返回空字节 (b'').

我实际上正在研究类似的东西并且 运行 处理相同的问题。我想到的解决方案是使用 the File option (not the UploadFile one),将输入写入另一个文件,然后执行解压缩。

这是我使用 TemporaryDirectory 和用于输入和解压缩文件的单独目录的实现:

@router.post("/test_file/unzip")
def unzip_upload(file: bytes = File(...)):

    with tempfile.TemporaryDirectory() as temp_dir:
        os.chdir(temp_dir)
        os.mkdir('input')
        os.mkdir('unzipped')

        with open("input/zip_file.zip", 'wb') as new_file:
            new_file.write(file)
            # print(f"output of listdir for /input {os.listdir(temp_dir + '/input')}")

            with zipfile.ZipFile("input/zip_file.zip") as zip_file:
                print(f"files in zip: {zip_file.namelist()}")
                zip_file.extractall('unzipped')

        unzipped_files = os.listdir('unzipped')

    return {"unzipped files": unzipped_files}

如果您需要使用 UploadFile 选项,我可以使用它:

@router.post("/test_uploadfile/unzip")
def unzip_upload(file: UploadFile = File(...)):

    with tempfile.TemporaryDirectory() as temp_dir:
        os.chdir(temp_dir)
        os.mkdir('input')
        os.mkdir('unzipped')

        with open("input/zip_file.zip", 'wb') as new_file:
            new_file.write(file.file._file.getvalue())
            # print(f"output of listdir for /input {os.listdir(temp_dir + '/input')}")

            with zipfile.ZipFile("input/zip_file.zip") as zip_file:
                print(f"files in zip: {zip_file.namelist()}")
                zip_file.extractall('unzipped')

        unzipped_files = os.listdir('unzipped')

    return {"unzipped files": unzipped_files}
@app.post("/", status_code=200)
async def upload(file: bytes = File(...)):
 zip_file_name = "file.zip"
    
 with open(zip_file_name, 'wb') as zip_file:
  zip_file.write(file)

我从接受 tempfile.TemporaryDirectory() 切换到 C H 建议的字节。