Как сохранить 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это просто оболочка SpooledTemporaryFile, к которой можно получить доступ как UploadFile.file.

Функция SpooledTemporaryFile () [...] работает точно так же, как TemporaryFile ()

Учитывая , для TemporaryFile:

Возвращает файловый объект, который можно использовать как область временного хранения. [..] Он будет уничтожен, как только он будет закрыт (включая неявное закрытие при сборке мусора). В Unix запись в каталоге для файла либо вообще не создается, либо удаляется сразу после создания файла. Другие платформы не поддерживают это; ваш код не должен полагаться на временный файл, созданный с помощью этой функции, имеющий или не имеющий видимого имени в файловой системе.

Вы должны использовать следующие асинхронные методы из UploadFile: write, read, seekи close. Они выполняются в пуле потоков и ожидаются асинхронно.

Обновление : Также я хотел бы процитировать несколько полезных служебных функций из этой темы (все кредиты @dmontagu), использующих shutil.copyfileobjс internal 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}'"}