Несколько практических вопросов по загрузке файлов в фастапи

Теги: python  fastapi 

How do I return an image in fastAPI

Если у вас уже есть байты изображения в памяти

fastapi.responses.Response вашими настройками content и media_type.

Вам также придется повозиться с декоратором эндпоинта, чтобы FastAPI указал правильный тип носителя в спецификации OpenAPI.

@app.get(
    "/image",

    # Set what the media type will be in the autogenerated OpenAPI specification.
    responses = {
        200: {
            "content": {"image/png": {}}
        }
    },

    # Prevent FastAPI from adding "application/json" as an additional
    # response media type in the autogenerated OpenAPI specification.
    # https://github.com/tiangolo/fastapi/issues/3258
    response_class=Response
)
def get_image()
    image_bytes: bytes = generate_cat_picture()
    # media_type here sets the media type of the actual response sent to the client.
    return Response(content=image_bytes, media_type="image/png")

Если ваше изображение существует только в файловой системе

fastapi.responses.FileResponse

документация

Будь осторожен с StreamingResponse

StreamingResponse труднее использовать правильно. StreamingResponse соответствует фрагментированному кодированию передачи HTTP . (Это может зависеть от вашего сервера ASGI, но , по крайней мере, это относится к Uvicorn .) И это не лучший вариант использования кодирования фрагментированной передачи. Этот ответ кодирует передачу по частям. Кодирование передачи по частям имеет смысл, когда вы не знаете заранее размер выходных данных и не хотите ждать, чтобы собрать все вместе, прежде чем начать отправлять его клиенту. Это может применяться к таким вещам, как обслуживание результатов медленных запросов к базе данных, но обычно не применимо к обслуживанию изображений. Ненужное фрагментированное кодирование передачи может быть вредным.

Источник

How to Upload File using FastAPI

Необходимо установить python-multipart. Необходимо внимательно отнестись к операциям функции, чтобы не столкнуться с блокировкой асинхронного цикла, если функция объявлена как асинхронная. Используемый SpooledTemporaryFile FastAPI/Starlette max_size атрибут имеет значение 1 МБ, что означает, что данные помещаются в память до тех пор, пока размер файла не превысит 1 МБ, после чего данные записываются во временный файл на диске во временном каталоге ОС. Следовательно, если вы загрузите файл размером более 1 МБ, он не будет сохранен в памяти, и вызов file.file.read() фактически считывает данные с диска в память. Таким образом, если файл слишком велик, чтобы поместиться в оперативную память вашего сервера, вам следует читать файл по частям и обрабатывать один фрагмент за раз, как описано в разделе «Чтение файла по частям». Вы также можете взглянуть на этот ответ , который демонстрирует другой подход к загрузке большого файла частями, используя метод Starlette.stream() и streaming-form-data пакет, позволяющий анализировать multipart/form-data фрагменты потоковой передачи, что позволяет значительно минимизировать время, необходимое для загрузки файлов.

Если вам необходимо определить свою конечную точку с помощью async def т.к. это может потребоваться await для некоторых других сопрограмм внутри вашего маршрута — тогда вам следует использовать асинхронное чтение и запись содержимого, как показано в этом ответе. Более того, если вам нужно отправить дополнительные данные (например, JSON данные) вместе с загрузкой файла (файлов), ознакомьтесь с этим ответом. Я бы также предложил вам взглянуть на этот ответ, который объясняет разницу между def и async def эндпоинтами.

Загрузка одного файла

from fastapi import File, UploadFile

@app.post("/upload")
def upload(file: UploadFile = File(...)):
    try:
        contents = file.file.read()
        with open(file.filename, 'wb') as f:
            f.write(contents)
    except Exception:
        return {"message": "There was an error uploading the file"}
    finally:
        file.file.close()

    return {"message": f"Successfully uploaded {file.filename}"}

или по частям (слишком большой файл)

from fastapi import File, UploadFile

@app.post("/upload")
def upload(file: UploadFile = File(...)):
    try:
        with open(file.filename, 'wb') as f:
            while contents := file.file.read(1024 * 1024):
                f.write(contents)
    except Exception:
        return {"message": "There was an error uploading the file"}
    finally:
        file.file.close()

    return {"message": f"Successfully uploaded {file.filename}"}

можно так-же через shutil.copyfileobj() (смотри в исходнике)

Загрузка нескольких файлов

from fastapi import File, UploadFile
from typing import List

@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
    for file in files:
        try:
            contents = file.file.read()
            with open(file.filename, 'wb') as f:
                f.write(contents)
        except Exception:
            return {"message": "There was an error uploading the file(s)"}
        finally:
            file.file.close()

    return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}

по частям

from fastapi import File, UploadFile
from typing import List

@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
    for file in files:
        try:
            with open(file.filename, 'wb') as f:
                while contents := file.file.read(1024 * 1024):
                    f.write(contents)
        except Exception:
            return {"message": "There was an error uploading the file(s)"}
        finally:
            file.file.close()

    return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}

Смотри еще: