Come salvare UploadFile in FastAPI

Question

Accetto il file tramite POST. Quando lo salvo in locale, posso leggere il contenuto utilizzando file.read (), ma viene visualizzato il nome tramite file.name errato (16). Quando provo a trovarlo con questo nome, ricevo un errore. Quale potrebbe essere il problema?

Il mio codice:

  @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

UploadFileè solo un wrapper intorno SpooledTemporaryFile, a cui è possibile accedere come UploadFile.file.

La funzione SpooledTemporaryFile () [...] opera esattamente come TemporaryFile ()

Dato per TemporaryFile:

Restituisce un oggetto simile a un file che può essere utilizzato come area di archiviazione temporanea. [..] Verrà distrutto non appena viene chiuso (inclusa una chiusura implicita quando l'oggetto viene sottoposto a garbage collection). In Unix, la voce di directory per il file non viene creata affatto o viene rimossa immediatamente dopo la creazione del file. Altre piattaforme non lo supportano; il codice non deve fare affidamento su un file temporaneo creato utilizzando questa funzione che ha o non ha un nome visibile nel file system.

Si consiglia di utilizzare le seguenti asincroni metodi di UploadFile: write, read, seeke close. Vengono eseguiti in un pool di thread e attesi in modo asincrono.

Aggiornamento : Inoltre, vorrei citare diverse utili funzioni di utilità da questo argomento (tutti i crediti @dmontagu) utilizzando shutil.copyfileobjcon interno 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 : ti consigliamo di utilizzare le funzioni di cui sopra all'interno degli defendpoint, non async def, poiché fanno uso di API di blocco.

Answer 2

Puoi salvare i file caricati in questo modo,

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}'"}

Questo è quasi identico all'uso del shutil.copyfileobj(...)metodo.

Quindi, la funzione sopra può essere riscritta come,

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}'"}