首页 > 解决方案 > 从 zip 的 SpooledTemporaryFile 获取信息

问题描述

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

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

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

标签: pythonfastapi

解决方案


我实际上正在做类似的事情并遇到了同样的问题。我想出的解决方案是使用选项File(不是那个UploadFile,将输入写入另一个文件,然后执行解压缩。

这是我使用 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}

推荐阅读