Cómo guardar UploadFile en FastAPI

Question

Acepto el archivo vía POST. Cuando lo guardo localmente, puedo leer el contenido usando file.read (), pero se muestra el nombre a través de file.name incorrecto (16). Cuando intento buscarlo con este nombre, aparece un error. ¿Cuál podría ser el problema?

Mi código:

  @router.post(
    path="/po/{id_po}/upload",
    response_model=schema.ContentUploadedResponse,
)
async def upload_file(
        id_po: int,
        background_tasks: BackgroundTasks,
        uploaded_file: UploadFile = File(...)):
    """pass"""
    uploaded_file.file.rollover()
    uploaded_file.file.flush()
    #shutil.copy(uploaded_file.file.name, f'/home/fyzzy/Desktop/api/{uploaded_file.filename}')
    background_tasks.add_task(s3_upload, uploaded_file=fp)
    return schema.ContentUploadedResponse()

Answer 1

UploadFilees solo un contenedor SpooledTemporaryFile, al que se puede acceder como UploadFile.file.

La función SpooledTemporaryFile () [...] opera exactamente como lo hace TemporaryFile ()

Dado para TemporaryFile:

Devuelve un objeto similar a un archivo que se puede utilizar como área de almacenamiento temporal. [..] Se destruirá tan pronto como se cierre (incluido un cierre implícito cuando el objeto sea recolectado como basura). En Unix, la entrada de directorio para el archivo no se crea en absoluto o se elimina inmediatamente después de que se crea el archivo. Otras plataformas no admiten esto; su código no debe depender de un archivo temporal creado con esta función que tenga o no un nombre visible en el sistema de archivos.

Debe utilizar las siguientes asincrónicos métodos de UploadFile: write, read, seeky close. Se ejecutan en un grupo de subprocesos y se esperan de forma asincrónica.

Actualización : Además, me gustaría citar varias funciones de utilidad útiles de este tema (todos los créditos @dmontagu) usando shutil.copyfileobjcon internal UploadFile.file:

import shutil
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Callable

from fastapi import UploadFile


def save_upload_file(upload_file: UploadFile, destination: Path) -> None:
    try:
        with destination.open("wb") as buffer:
            shutil.copyfileobj(upload_file.file, buffer)
    finally:
        upload_file.file.close()


def save_upload_file_tmp(upload_file: UploadFile) -> Path:
    try:
        suffix = Path(upload_file.filename).suffix
        with NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
            shutil.copyfileobj(upload_file.file, tmp)
            tmp_path = Path(tmp.name)
    finally:
        upload_file.file.close()
    return tmp_path


def handle_upload_file(
    upload_file: UploadFile, handler: Callable[[Path], None]
) -> None:
    tmp_path = save_upload_file_tmp(upload_file)
    try:
        handler(tmp_path)  # Do something with the saved temp file
    finally:
        tmp_path.unlink()  # Delete the temp file

Nota : querrá usar las funciones anteriores dentro de los defpuntos finales, no async def, ya que hacen uso de API de bloqueo.

Answer 2

Puede guardar los archivos cargados de esta manera,

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/upload-file/")
async def create_upload_file(uploaded_file: UploadFile = File(...)):
    file_location = f"files/{uploaded_file.filename}" with open(file_location, "wb+") as file_object: file_object.write(uploaded_file.file.read())
    return {"info": f"file '{uploaded_file.filename}' saved at '{file_location}'"}

Esto es casi idéntico al uso del shutil.copyfileobj(...)método.

Entonces, la función anterior se puede reescribir como,

import shutil
from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/upload-file/")
async def create_upload_file(uploaded_file: UploadFile = File(...)):    
file_location = f"files/{uploaded_file.filename}"
    with open(file_location, "wb+") as file_object:
        shutil.copyfileobj(uploaded_file.file, file_object)    
return {"info": f"file '{uploaded_file.filename}' saved at '{file_location}'"}