So speichern Sie UploadFile in FastAPI

Question

Ich akzeptiere die Datei per POST. Wenn ich es lokal speichere, kann ich den Inhalt mit file.read () lesen, aber der Name über file.name false (16) wird angezeigt. Wenn ich versuche, es unter diesem Namen zu finden, wird eine Fehlermeldung angezeigt. Was könnte das Problem sein?

Mein 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

UploadFileist nur ein Wrapper herum SpooledTemporaryFile, auf den als zugegriffen werden kann UploadFile.file.

Die Funktion SpooledTemporaryFile () [...] funktioniert genau wie TemporaryFile ()

Gegeben für TemporaryFile:

Gibt ein dateiähnliches Objekt zurück, das als temporärer Speicherbereich verwendet werden kann. [..] Es wird zerstört, sobald es geschlossen wird (einschließlich eines impliziten Schließens, wenn das Objekt durch Müll gesammelt wird). Unter Unix wird der Verzeichniseintrag für die Datei entweder gar nicht erstellt oder unmittelbar nach dem Erstellen der Datei entfernt. Andere Plattformen unterstützen dies nicht. Ihr Code sollte nicht auf einer temporären Datei basieren, die mit dieser Funktion erstellt wurde und einen sichtbaren Namen im Dateisystem hat oder nicht.

Sie sollten die folgenden async verwenden Methoden der Upload: write, read, seekund close. Sie werden in einem Thread-Pool ausgeführt und asynchron erwartet.

Update : Außerdem möchte ich einige nützliche Dienstprogrammfunktionen aus diesem Thema (alle Credits @dmontagu) shutil.copyfileobjmit intern zitieren 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

Hinweis : Sie möchten die oben genannten Funktionen nicht innerhalb von defEndpunkten verwenden async def, da diese blockierende APIs verwenden.

Answer 2

Sie können die hochgeladenen Dateien auf diese Weise speichern.

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

Dies ist fast identisch mit der Verwendung der shutil.copyfileobj(...)Methode.

Die obige Funktion kann also wie folgt umgeschrieben werden:

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