FastAPI'de UploadFile nasıl kaydedilir?

Question

Dosyayı POST ile kabul ediyorum. Yerel olarak kaydettiğimde, içeriği file.read () kullanarak okuyabiliyorum, ancak file.name yoluyla ad yanlış (16) görüntüleniyor. Bu isimle bulmaya çalıştığımda bir hata alıyorum. Sorun ne olabilir?

Kodum:

  @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

UploadFilesadece bir sarmalayıcıdır SpooledTemporaryFile, olarak erişilebilir UploadFile.file.

SpooledTporaryFile () [...] işlevi tam olarak TemporaryFile () işlevinin yaptığı gibi çalışır

Verilen için TemporaryFile:

Bir İade dosyası benzeri bir geçici depolama alanı olarak kullanılabilir nesneyi. [..] Kapanır kapanmaz yok edilecektir (nesne çöp toplandığında örtük bir kapatma dahil). Unix altında, dosya için dizin girişi ya hiç oluşturulmaz ya da dosya oluşturulduktan hemen sonra kaldırılır. Diğer platformlar bunu desteklemez; Kodunuz, dosya sisteminde görünür bir ada sahip olan veya olmayan bu işlev kullanılarak oluşturulmuş geçici bir dosyaya dayanmamalıdır.

Aşağıdaki zaman uyumsuz kullanması gereken yöntemleri DosyaYükle ait: write, read, seekve close. Bir iş parçacığı havuzunda çalıştırılırlar ve zaman uyumsuz olarak beklenirler.

Güncelleme : Ayrıca, dahili olarak kullanarak bu konudaki (tüm krediler @dmontagu) birkaç yararlı yardımcı program işlevinden alıntı yapmak istiyorum :shutil.copyfileobjUploadFile.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

Not : Yukarıdaki işlevleri , engelleme API'lerinden yararlandıkları için defuç noktaların içinde kullanmak istemezsiniz async def.

Answer 2

Yüklenen dosyaları bu şekilde kaydedebilirsiniz,

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

Bu, shutil.copyfileobj(...)yöntemin kullanımıyla neredeyse aynıdır .

Dolayısıyla, yukarıdaki işlev şu şekilde yeniden yazılabilir:

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