Загрузка файлов в FastAPI: готовое решение за 20 минут

Что нужно до начала

Убедитесь, что у вас есть:

• Python 3.10 или новее;
• установленный FastAPI 0.115.x и Uvicorn (pip install "fastapi[standard]" ставит оба);
• библиотека python-multipart — без неё FastAPI не сможет разбирать multipart/form-data. Если ставили fastapi[standard], она уже есть; иначе — pip install python-multipart;
• каталог, куда будут сохраняться файлы, с правами на запись для процесса.

В результате вы получите эндпоинт, который принимает файл по HTTP, проверяет размер и MIME-тип, записывает на диск потоково (без загрузки целиком в память) и возвращает метаданные в JSON.

Шаг 1 — минимальный эндпоинт

Начнём с самого короткого варианта, чтобы убедиться, что инфраструктура работает.

# app.py — FastAPI 0.115.x, Python 3.10+
from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile):
    contents = await file.read()
    return {
        "filename": file.filename,
        "size_bytes": len(contents),
        "content_type": file.content_type,
    }

Запуск:

uvicorn app:app --reload

Проверка: отправьте файл через curl:

curl -F "file=@photo.jpg" http://127.0.0.1:8000/upload

В ответе должен прийти JSON с именем файла, размером и типом. Если видите ошибку 422 Unprocessable Entity — скорее всего, не установлена python-multipart.

Этот вариант рабочий, но опасный: await file.read() загружает весь файл в оперативную память. Для аватарки в 200 КБ это терпимо, для видео в 500 МБ — нет.

Шаг 2 — потоковая запись на диск

Чтобы не держать файл в памяти, читаем его чанками и пишем на диск по мере поступления.

# app.py — потоковая загрузка
import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

CHUNK_SIZE = 1024 * 1024  # 1 МБ

@app.post("/upload")
async def upload_file(file: UploadFile):
    dest = UPLOAD_DIR / file.filename

    # Защита от перезаписи
    if dest.exists():
        raise HTTPException(status_code=409, detail="Файл с таким именем уже существует")

    total_size = 0
    try:
        with dest.open("wb") as out:
            while chunk := await file.read(CHUNK_SIZE):
                total_size += len(chunk)
                out.write(chunk)
    except Exception:
        dest.unlink(missing_ok=True)  # удаляем недозаписанный файл
        raise

    return {
        "filename": file.filename,
        "size_bytes": total_size,
        "path": str(dest),
    }

Проверка: отправьте файл размером больше 10 МБ. Память процесса Uvicorn не должна расти пропорционально размеру файла — можно посмотреть через htop или psutil.

Почему именно чанки, а не shutil.copyfileobj

shutil.copyfileobj работает с синхронным файловым объектом. UploadFile в FastAPI — обёртка над SpooledTemporaryFile из Starlette, и его метод read() — корутина. Если вы хотите использовать shutil.copyfileobj, сначала нужно получить внутренний файловый объект через file.file, но тогда вы теряете контроль над размером и асинхронность.

Шаг 3 — валидация размера и типа

Принимать любой файл любого размера — приглашение к злоупотреблению. Ограничим оба параметра.

# app.py — с валидацией
import uuid
from pathlib import Path
from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 МБ
ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp", "application/pdf"}
CHUNK_SIZE = 1024 * 1024

@app.post("/upload")
async def upload_file(file: UploadFile):
    # Проверка MIME-типа (передаётся клиентом — не доверяем полностью)
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=415,
            detail=f"Тип {file.content_type} не разрешён. Допустимы: {', '.join(ALLOWED_TYPES)}",
        )

    # Генерируем уникальное имя, чтобы избежать коллизий
    suffix = Path(file.filename).suffix
    safe_name = f"{uuid.uuid4().hex}{suffix}"
    dest = UPLOAD_DIR / safe_name

    total_size = 0
    try:
        with dest.open("wb") as out:
            while chunk := await file.read(CHUNK_SIZE):
                total_size += len(chunk)
                if total_size > MAX_FILE_SIZE:
                    raise HTTPException(
                        status_code=413,
                        detail=f"Файл превышает лимит в {MAX_FILE_SIZE // (1024*1024)} МБ",
                    )
                out.write(chunk)
    except HTTPException:
        dest.unlink(missing_ok=True)
        raise
    except Exception:
        dest.unlink(missing_ok=True)
        raise

    return {
        "original_name": file.filename,
        "saved_as": safe_name,
        "size_bytes": total_size,
        "content_type": file.content_type,
    }

Проверка: попробуйте отправить .exe-файл — получите 415. Попробуйте файл больше 10 МБ — получите 413, а недозаписанный файл удалится с диска.

Заголовок Content-Type, который передаёт клиент, — не доказательство. Браузер определяет его по расширению, curl — по таблице MIME. Для серьёзной валидации нужно читать magic bytes файла (первые несколько байт). Библиотека python-magic делает это, но требует libmagic в системе.

Шаг 4 — приём нескольких файлов

Достаточно изменить сигнатуру: принимать список UploadFile.

from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/upload-multiple")
async def upload_multiple(files: list[UploadFile]):
    results = []
    for file in files:
        contents = await file.read()
        results.append({"filename": file.filename, "size": len(contents)})
    return results

curl для нескольких файлов:

curl -F "files=@a.jpg" -F "files=@b.png" http://127.0.0.1:8000/upload-multiple

Обратите внимание: имя параметра в curl (files) должно совпадать с именем аргумента функции.

Сравнение подходов

^* Операция out.write(chunk) — синхронный ввод-вывод. При больших нагрузках стоит вынести запись в run_in_executor или использовать aiofiles. Для умеренного трафика разница незаметна.

Если не получилось

1. 422 Unprocessable Entity при отправке файла. Почти всегда — не установлена python-multipart. Установите: pip install python-multipart и перезапустите сервер.
2. 400 Bad Request без подробностей. Проверьте, что в curl имя поля (file) совпадает с именем аргумента в функции. Несовпадение — самая частая причина.
3. Файл сохраняется с нулевым размером. Вы уже вызвали await file.read() раньше в коде, и указатель стоит в конце. Перед повторным чтением нужно await file.seek(0).
4. PermissionError при записи. У процесса Uvicorn нет прав на запись в каталог. Проверьте владельца и права: ls -la uploads/.
5. Большой файл обрывается без ошибки. Проверьте настройки reverse proxy. Nginx по умолчанию ограничивает тело запроса в 1 МБ — директива client_max_body_size.

Сноска: UploadFile — класс Starlette (FastAPI его реэкспортирует), обёртка над SpooledTemporaryFile. Файлы до определённого порога (по умолчанию 1 МБ) хранятся в памяти, свыше — во временном файле на диске. Это поведение SpooledTemporaryFile из стандартной библиотеки Python, и порог задаётся Starlette при создании объекта.

Что делать дальше

• Для продакшена поставьте Nginx перед Uvicorn и настройте client_max_body_size в соответствии с вашим лимитом.
• Если файлы хранятся долго — добавьте запись метаданных в базу данных (имя, размер, дата, кто загрузил) и отдачу по ID вместо прямого пути.
• Для файлов свыше 50–100 МБ рассмотрите presigned URL в S3-совместимое хранилище: файл пойдёт напрямую от клиента в хранилище, минуя ваш сервер.
• Измерьте пиковое потребление памяти под нагрузкой — tracemalloc из стандартной библиотеки покажет, сколько съедает каждый запрос.