Estoy cargando un archivo zip con Fast API y toma los archivos como SpooledTemporaryFile. He estado tratando de copiar ese archivo en el disco. He intentado varias cosas; los dos en los que siento que he progresado están abajo.
Intenté usar zipfile.Zipfile, cuando lo convierto en uno no puedo descomprimirlo porque dice que no es un zip.
También cargué un solo archivo que no está comprimido y convertí SpooledTemporaryFile en _io.BytesIO, sin embargo, cuando intento leer el contenido de un solo archivo (no comprimido) devolvió un byte vacío (b'').
De hecho, estoy trabajando en algo similar y me encontré con los mismos problemas. La solución que se me ocurrió fue usar la opción File (no la de UploadFile ) , escribir la entrada en otro archivo y luego realizar la descompresión.
Aquí está mi implementación usando un directorio temporal y directorios separados para archivos de entrada y descomprimidos:
@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} Si necesita usar la opción UploadFile , conseguí que esto funcionara:
@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}