วิธีบันทึก UploadFile ใน FastAPI

Question

ฉันยอมรับไฟล์ผ่าน POST เมื่อฉันบันทึกในเครื่องฉันสามารถอ่านเนื้อหาโดยใช้ file.read () แต่ชื่อผ่าน file.name ไม่ถูกต้อง (16) แสดงขึ้น เมื่อฉันพยายามค้นหาโดยใช้ชื่อนี้ฉันได้รับข้อผิดพลาด ปัญหาอาจเกิดจากอะไร?

รหัสของฉัน:

  @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เป็นเพียงห่อหุ้มรอบซึ่งสามารถเข้าถึงได้เป็นSpooledTemporaryFileUploadFile.file

SpooledTem ContemporaryFile () [... ] ฟังก์ชันทำงานเหมือนกับ TemporaryFile () ทำ

มอบให้สำหรับTemporaryFile:

ส่งคืนวัตถุคล้ายไฟล์ที่สามารถใช้เป็นพื้นที่จัดเก็บชั่วคราว [.. ] มันจะถูกทำลายทันทีที่ปิด (รวมถึงการปิดโดยปริยายเมื่อวัตถุถูกเก็บขยะ) ภายใต้ Unix รายการไดเร็กทอรีสำหรับไฟล์จะไม่ถูกสร้างขึ้นเลยหรือถูกลบออกทันทีหลังจากสร้างไฟล์ แพลตฟอร์มอื่นไม่รองรับสิ่งนี้ โค้ดของคุณไม่ควรอาศัยไฟล์ชั่วคราวที่สร้างขึ้นโดยใช้ฟังก์ชันนี้โดยมีหรือไม่มีชื่อที่มองเห็นได้ในระบบไฟล์

คุณควรใช้ async ต่อไปนี้วิธีการของ UploadFile: write, read, และseek closeพวกเขาจะดำเนินการในเธรดพูลและรอแบบอะซิงโครนัส

อัปเดต : นอกจากนี้ฉันต้องการอ้างถึงฟังก์ชันยูทิลิตี้ที่มีประโยชน์หลายอย่างจากหัวข้อนี้(เครดิตทั้งหมด @dmontagu) โดยใช้shutil.copyfileobjกับภายใน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

หมายเหตุ : คุณต้องการใช้ฟังก์ชันข้างต้นภายในdefจุดสิ้นสุดไม่ใช่async defเนื่องจากใช้ประโยชน์จาก API การบล็อก

Answer 2

คุณสามารถบันทึกไฟล์ที่อัปโหลดด้วยวิธีนี้

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

เกือบจะเหมือนกับการใช้shutil.copyfileobj(...)วิธีการ

ดังนั้นฟังก์ชันข้างต้นสามารถเขียนใหม่ได้เป็น

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