FastAPIMiddleCoding

Как обрабатывать загрузку файлов (file uploads) в FastAPI?

Используйте UploadFile для потоковой загрузки файлов, читайте чанками через await file.read(chunk_size), проверяйте MIME через python-magic и не доверяйте Content-Type от клиента.

Как работает загрузка файлов в FastAPI

FastAPI предоставляет два инструмента для приёма файлов: File (читает данные в память) и UploadFile (потоковая обработка через SpooledTemporaryFile). Для большинства случаев нужен именно UploadFile — он не грузит всё в RAM сразу.

Одиночная загрузка

from fastapi import FastAPI, File, UploadFile, HTTPException
from pathlib import Path
import aiofiles
import uuid

app = FastAPI()
UPLOAD_DIR = Path("/tmp/uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload/")
async def upload_file(file: UploadFile) -> dict:
    # Проверяем MIME-тип
    if file.content_type not in ("image/jpeg", "image/png", "application/pdf"):
        raise HTTPException(status_code=415, detail="Unsupported media type")

    dest = UPLOAD_DIR / f"{uuid.uuid4()}_{file.filename}"
    async with aiofiles.open(dest, "wb") as out:
        while chunk := await file.read(1024 * 64):  # 64 KB chunks
            await out.write(chunk)

    return {"filename": file.filename, "saved_as": dest.name, "size": dest.stat().st_size}

Множественная загрузка

from typing import Annotated

@app.post("/upload/multiple/")
async def upload_files(
    files: Annotated[list[UploadFile], File(description="Up to 5 files")],
) -> list[dict]:
    if len(files) > 5:
        raise HTTPException(status_code=400, detail="Too many files")
    results = []
    for f in files:
        content = await f.read()
        results.append({"filename": f.filename, "bytes": len(content)})
    return results

Форма с файлом и текстовыми полями

Если нужно принять и файл, и JSON-поля одновременно, используется Form. При этом нельзя объявить тело запроса как Pydantic-модель — только отдельные Form()-параметры.

from fastapi import Form

@app.post("/profile/")
async def update_profile(
    username: Annotated[str, Form()],
    avatar: UploadFile,
) -> dict:
    data = await avatar.read()
    return {"username": username, "avatar_size": len(data)}

Ограничение размера через middleware

FastAPI сам по себе не ограничивает размер тела запроса. Добавьте middleware или проверяйте через Content-Length:

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

MAX_UPLOAD_BYTES = 10 * 1024 * 1024  # 10 MB

class LimitUploadSizeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.method == "POST":
            cl = request.headers.get("content-length")
            if cl and int(cl) > MAX_UPLOAD_BYTES:
                return Response("File too large", status_code=413)
        return await call_next(request)

app.add_middleware(LimitUploadSizeMiddleware)

Загрузка в S3/MinIO

import aioboto3

s3 = aioboto3.Session()

@app.post("/upload/s3/")
async def upload_to_s3(file: UploadFile) -> dict:
    async with s3.client(
        "s3",
        endpoint_url="http://minio:9000",
        aws_access_key_id="minioadmin",
        aws_secret_access_key="minioadmin",
    ) as client:
        await client.upload_fileobj(
            file.file,
            "my-bucket",
            f"uploads/{uuid.uuid4()}_{file.filename}",
            ExtraArgs={"ContentType": file.content_type or "application/octet-stream"},
        )
    return {"status": "uploaded"}

Подводные камни

  • await file.read() без аргумента загружает весь файл в память — для больших файлов (видео, архивы) используйте чтение чанками в цикле.
  • Повторный вызов file.read() вернёт пустые байты — файловый указатель уже в конце. Перед повторным чтением вызывайте await file.seek(0).
  • Проверка MIME-типа через file.content_type основана на заголовке Content-Type от клиента — его можно подделать. Для надёжной валидации используйте библиотеку python-magic (libmagic) по реальному содержимому файла.
  • При работе в нескольких воркерах Gunicorn/Uvicorn сохранять файлы в локальную файловую систему ненадёжно — разные воркеры видят разные директории. Используйте общее хранилище (S3, MinIO, NFS).
  • Зависимость python-multipart должна быть установлена явно; без неё FastAPI выбросит ошибку при первом обращении к Form/File.
  • Content-Type запроса должен быть multipart/form-data — если клиент отправляет application/json, файл не придёт.
  • SpooledTemporaryFile (внутри UploadFile) по умолчанию переключается на диск при превышении 1 MB — настраивается параметром spool_max_size у Starlette.
  • Имя файла file.filename приходит от пользователя и может содержать path traversal (../../etc/passwd). Всегда санируйте имя через pathlib.Path(file.filename).name или генерируйте новое UUID-имя.

Common mistakes

  • Описывать file uploads только как термин и не показывать механизм на минимальном примере.
  • Игнорировать ошибки, пустые данные, конкурентный доступ или границы транзакции.
  • Не связывать поведение с официальным контрактом FastAPI и реальной эксплуатацией.

What the interviewer is testing

  • Объясняет file uploads через последовательность действий, а не через набор ключевых слов.
  • Приводит короткий кодовый пример или production-сценарий с ожидаемым поведением.
  • Называет хотя бы один риск: производительность, безопасность, транзакции, память или сопровождение.

Sources

Related topics