Comment enregistrer UploadFile dans FastAPI

Question

J'accepte le fichier via POST. Quand je l'enregistre localement, je peux lire le contenu en utilisant file.read (), mais le nom via file.name incorrect (16) est affiché. Lorsque j'essaie de le trouver par ce nom, j'obtiens une erreur. Quel pourrait être le problème?

Mon code:

  @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

UploadFileest juste un wrapper SpooledTemporaryFile, accessible en tant que UploadFile.file.

La fonction SpooledTemporaryFile () [...] fonctionne exactement comme TemporaryFile ()

Donné pour TemporaryFile:

Renvoie un objet de type fichier qui peut être utilisé comme zone de stockage temporaire. [..] Il sera détruit dès sa fermeture (y compris une fermeture implicite lorsque l'objet est ramassé). Sous Unix, l'entrée de répertoire du fichier n'est pas du tout créée ou est supprimée immédiatement après la création du fichier. D'autres plates-formes ne prennent pas en charge cela; votre code ne doit pas reposer sur un fichier temporaire créé à l'aide de cette fonction ayant ou n'ayant pas de nom visible dans le système de fichiers.

Vous devez utiliser les suivantes async méthodes de UploadFile: write, read, seeket close. Ils sont exécutés dans un pool de threads et attendus de manière asynchrone.

Mise à jour : Aussi, je voudrais citer plusieurs fonctions utilitaires utiles de cette rubrique (tous les crédits @dmontagu) en utilisant shutil.copyfileobjavec interne 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

Remarque : vous voudrez utiliser les fonctions ci-dessus à l'intérieur des defpoints de terminaison, non async def, car elles utilisent des API de blocage.

Answer 2

Vous pouvez enregistrer les fichiers téléchargés de cette façon,

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

C'est presque identique à l'utilisation de la shutil.copyfileobj(...)méthode.

Ainsi, la fonction ci-dessus peut être réécrite comme suit:

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