Cách lưu UploadFile trong FastAPI

Question

Tôi chấp nhận tệp qua POST. Khi tôi lưu nó cục bộ, tôi có thể đọc nội dung bằng file.read (), nhưng tên qua file.name không chính xác (16) được hiển thị. Khi tôi cố gắng tìm nó bằng tên này, tôi gặp lỗi. Rắc rối có thể là cái gì?

Mã của tôi:

  @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

UploadFilechỉ là một trình bao bọc xung quanh SpooledTemporaryFile, có thể được truy cập dưới dạng UploadFile.file.

Hàm SpooledTemporaryFile () [...] hoạt động chính xác như TemporaryFile ()

Đưa ra cho TemporaryFile:

Trả về một đối tượng giống tệp có thể được sử dụng làm vùng lưu trữ tạm thời. [..] Nó sẽ bị phá hủy ngay sau khi nó được đóng lại (bao gồm cả việc đóng ngầm khi đối tượng được thu gom rác). Trong Unix, mục nhập thư mục cho tệp hoàn toàn không được tạo hoặc bị xóa ngay sau khi tệp được tạo. Các nền tảng khác không hỗ trợ điều này; mã của bạn không nên dựa vào tệp tạm thời được tạo bằng chức năng này có hoặc không có tên hiển thị trong hệ thống tệp.

Bạn nên sử dụng async sau phương pháp của UploadFile: write, read, seekvà close. Chúng được thực thi trong một nhóm luồng và được chờ đợi không đồng bộ.

Cập nhật : Ngoài ra, tôi muốn trích dẫn một số chức năng tiện ích hữu ích từ chủ đề này (tất cả các khoản tín dụng @dmontagu) bằng cách sử dụng shutil.copyfileobjvới nội bộ 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

Lưu ý : bạn không muốn sử dụng các chức năng trên bên trong các defđiểm cuối async def, vì chúng sử dụng các API chặn.

Answer 2

Bạn có thể lưu các tệp đã tải lên theo cách này,

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

Điều này gần giống với cách sử dụng shutil.copyfileobj(...)method.

Vì vậy, hàm trên có thể được viết lại thành,

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