Bài 7: Upload file (image, PDF) và xử lý

Sau bài này bạn sẽ:

• ✅ Phân biệt File(...) và UploadFile — khi nào dùng cái nào
• ✅ Viết endpoint nhận ảnh, validate content-type, đưa vào Pillow/PyTorch để classify
• ✅ Nhận batch nhiều file trong một request và xử lý batch inference
• ✅ Kết hợp metadata (form field) với file trong cùng một request multipart
• ✅ Nhận PDF, parse text bằng pypdf để đưa vào RAG pipeline
• ✅ Đọc file lớn theo chunk, tránh OOM
• ✅ Giới hạn kích thước file qua middleware hoặc reverse proxy
• ✅ Biết các pitfall hay gặp khi làm việc với UploadFile

FastAPI cung cấp hai cách khai báo file trong endpoint:

File(...) — bytes trực tiếp

from fastapi import FastAPI, File

app = FastAPI()

@app.post("/upload-bytes")
async def upload_bytes(data: bytes = File(...)):
    # data đã là bytes — toàn bộ nội dung file nằm trong RAM
    return {"size": len(data)}

FastAPI load toàn bộ nội dung file vào RAM trước khi hàm handler chạy. Phù hợp cho file nhỏ (dưới vài MB) khi code đơn giản là ưu tiên. Với file 100 MB thì toàn bộ 100 MB nằm trong RAM ngay từ đầu.

UploadFile — SpooledTemporaryFile

from fastapi import FastAPI, UploadFile

app = FastAPI()

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

UploadFile bọc một SpooledTemporaryFile từ thư viện chuẩn Python. Hoạt động như sau:

• Khi file nhỏ hơn ngưỡng (mặc định 1 MB): dữ liệu nằm trong RAM.
• Khi file vượt ngưỡng: dữ liệu được spill (tràn) xuống file tạm trên disk. RAM không tăng thêm.

Ngoài .read(), UploadFile còn có:

• file.filename — tên file gốc từ client (string, có thể là None nếu client không gửi)
• file.content_type — MIME type từ header Content-Type của part (ví dụ "image/jpeg")
• await file.read(size=-1) — đọc toàn bộ (hoặc size byte) từ con trỏ hiện tại
• await file.seek(offset) — di chuyển con trỏ đến vị trí offset
• await file.close() — giải phóng SpooledTemporaryFile

Kết luận: Mặc định nên dùng UploadFile. Chỉ dùng File(...): bytes khi bạn chắc file luôn nhỏ và muốn code gọn hơn.

Cài đặt dependencies

pip install fastapi uvicorn[standard] pillow pypdf python-multipart

python-multipart bắt buộc để FastAPI xử lý được multipart/form-data. Nếu quên cài, FastAPI báo lỗi 422 Unprocessable Entity ngay cả khi code đúng.

Ví dụ đầy đủ kết hợp với ResNet được load qua lifespan (bài 5) và dependency injection:

import io
from contextlib import asynccontextmanager
from fastapi import FastAPI, UploadFile, Depends, HTTPException
from PIL import Image
import torch
import torchvision.transforms as T
import torchvision.models as models

# Transform chuẩn ImageNet
_transform = T.Compose([
    T.Resize(256),
    T.CenterCrop(224),
    T.ToTensor(),
    T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

_models: dict = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    resnet = models.resnet50(weights=models.ResNet50_Weights.DEFAULT)
    resnet.eval()
    _models["resnet"] = resnet
    yield
    _models.clear()

app = FastAPI(lifespan=lifespan)

ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/webp"}

def get_resnet():
    return _models["resnet"]

@app.post("/classify")
async def classify(
    file: UploadFile,
    model: torch.nn.Module = Depends(get_resnet),
):
    # Validate content-type
    if file.content_type not in ALLOWED_CONTENT_TYPES:
        raise HTTPException(
            status_code=415,
            detail=f"Unsupported media type: {file.content_type}. "
                   f"Accepted: {ALLOWED_CONTENT_TYPES}",
        )

    contents = await file.read()   # đọc bytes từ SpooledTemporaryFile

    try:
        img = Image.open(io.BytesIO(contents)).convert("RGB")
    except Exception:
        raise HTTPException(status_code=400, detail="Cannot decode image.")

    tensor = _transform(img).unsqueeze(0)  # (1, 3, 224, 224)

    with torch.no_grad():
        logits = model(tensor)
        probs = torch.softmax(logits, dim=1)
        top5_probs, top5_idx = torch.topk(probs, k=5)

    return {
        "top5": [
            {"class_id": idx.item(), "prob": round(p.item(), 4)}
            for idx, p in zip(top5_idx[0], top5_probs[0])
        ]
    }

Tại sao await file.read()?

UploadFile.read() là coroutine vì nó delegate xuống SpooledTemporaryFile qua thread pool (Starlette dùng run_in_threadpool ở bên trong). Nếu bỏ await, bạn nhận về một coroutine object, không phải bytes, và code tiếp theo sẽ crash.

Tại sao validate content-type?

Nếu endpoint nhận JPEG nhưng nhận được một file TIFF hoặc SVG, Pillow có thể raise exception hoặc xử lý sai. Trả ngay 415 Unsupported Media Type sớm rõ ràng hơn là để Pillow crash bên trong và trả về 500 chung chung.

Lưu ý: validate content_type là cần nhưng chưa đủ cho production. Xem thêm mục "Common pitfalls" ở cuối bài về magic byte check.

Khai báo tham số kiểu list[UploadFile] để nhận nhiều file trong một request. Client gửi cùng field name nhiều lần trong form-data.

from fastapi import FastAPI, UploadFile, Depends, HTTPException
import io, torch
from PIL import Image

app = FastAPI()  # giả sử lifespan đã có như ví dụ trước

ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/webp"}

@app.post("/classify-batch")
async def classify_batch(
    files: list[UploadFile],
    model: torch.nn.Module = Depends(get_resnet),
):
    if not files:
        raise HTTPException(status_code=400, detail="No files received.")
    if len(files) > 32:
        raise HTTPException(status_code=400, detail="Max 32 files per request.")

    tensors = []
    for f in files:
        if f.content_type not in ALLOWED_CONTENT_TYPES:
            raise HTTPException(
                status_code=415,
                detail=f"{f.filename}: unsupported type {f.content_type}",
            )
        contents = await f.read()
        img = Image.open(io.BytesIO(contents)).convert("RGB")
        tensors.append(_transform(img))   # (3, 224, 224)

    batch = torch.stack(tensors)           # (N, 3, 224, 224)

    with torch.no_grad():
        logits = model(batch)
        probs = torch.softmax(logits, dim=1)
        top1_idx = torch.argmax(probs, dim=1)

    return {
        "results": [
            {
                "filename": f.filename,
                "class_id": idx.item(),
                "prob": round(probs[i, idx].item(), 4),
            }
            for i, (f, idx) in enumerate(zip(files, top1_idx))
        ]
    }

Tại sao gom batch trước khi forward?

Gọi model(batch) một lần với tensor (N, 3, 224, 224) nhanh hơn nhiều so với gọi model(single_tensor) N lần riêng lẻ. GPU xử lý song song N ảnh cùng một kernel call; N lần forward riêng tốn N kernel launch overhead.

Gửi nhiều file từ client

# curl — dùng -F "files=@..." lặp lại nhiều lần cùng field name
curl -X POST \
  -F "[email protected]" \
  -F "[email protected]" \
  -F "[email protected]" \
  http://localhost:8000/classify-batch

Khi cần gửi thêm metadata cùng với file — ví dụ version model cần dùng, ID người dùng, hay ngưỡng confidence — dùng Form(...) cho từng field text:

from fastapi import FastAPI, UploadFile, Form, Depends, HTTPException
import io
from PIL import Image

app = FastAPI()

@app.post("/classify")
async def classify(
    file: UploadFile,
    model_version: str = Form(...),
    threshold: float = Form(0.5),
    model=Depends(get_resnet),
):
    if file.content_type not in ALLOWED_CONTENT_TYPES:
        raise HTTPException(status_code=415, detail="Unsupported file type.")

    contents = await file.read()
    img = Image.open(io.BytesIO(contents)).convert("RGB")
    tensor = _transform(img).unsqueeze(0)

    with torch.no_grad():
        probs = torch.softmax(model(tensor), dim=1)
        top1_prob, top1_idx = probs.max(dim=1)

    if top1_prob.item() < threshold:
        return {"class_id": None, "prob": round(top1_prob.item(), 4), "below_threshold": True}

    return {
        "model_version": model_version,
        "class_id": top1_idx.item(),
        "prob": round(top1_prob.item(), 4),
        "below_threshold": False,
    }

Gửi bằng curl:

curl -X POST \
  -F "[email protected]" \
  -F "model_version=v2" \
  -F "threshold=0.7" \
  http://localhost:8000/classify

Quan trọng: Khi endpoint có cả UploadFile lẫn field text, không dùng Body(...) cho field text. FastAPI sẽ báo lỗi vì Body() yêu cầu request body kiểu JSON, nhưng multipart request không phải JSON body. Tất cả field non-file phải dùng Form(...).

Nếu bạn muốn gửi JSON phức tạp kèm file, cách đơn giản nhất là serialize JSON thành string và gửi qua một form field, rồi parse lại trong handler:

import json
from fastapi import Form

@app.post("/classify-with-meta")
async def classify_with_meta(
    file: UploadFile,
    meta_json: str = Form(...),   # client gửi JSON.stringify(...)
):
    meta = json.loads(meta_json)
    ...

Endpoint điển hình cho RAG (Retrieval-Augmented Generation): nhận PDF, extract text, chuẩn bị đưa vào embedding + vector store.

import io
from fastapi import FastAPI, UploadFile, HTTPException
from pypdf import PdfReader

app = FastAPI()

@app.post("/ingest")
async def ingest_pdf(file: UploadFile):
    if file.content_type != "application/pdf":
        raise HTTPException(
            status_code=415,
            detail=f"Expected application/pdf, got {file.content_type}",
        )

    contents = await file.read()

    try:
        reader = PdfReader(io.BytesIO(contents))
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Cannot parse PDF: {e}")

    pages_text: list[str] = []
    for page in reader.pages:
        text = page.extract_text() or ""
        pages_text.append(text)

    full_text = "\n\n".join(pages_text)

    # Tiếp theo: chunk text → embed → upsert vào vector DB
    # (xem Module 3 — Vector Database)
    return {
        "filename": file.filename,
        "num_pages": len(reader.pages),
        "char_count": len(full_text),
        # Trong thực tế: trả về chunk IDs đã ingest thay vì full_text
        "preview": full_text[:500],
    }

pypdf 4.x (phát hành 2024) là fork tiếp nối của PyPDF2. Import đổi thành from pypdf import PdfReader. API page.extract_text() trả về string hoặc None nếu page không có text layer (ví dụ PDF scan hình ảnh).

Alternatives cho PDF parsing

• pdfplumber: extract bảng (table) tốt hơn pypdf. Trả về dữ liệu có tọa độ từng ký tự, phù hợp khi PDF có nhiều bảng tài chính, hóa đơn.
• unstructured (unstructured-io): xử lý layout phức tạp, tự nhận biết header/footer/table/list. Nặng hơn (nhiều dependency) nhưng output sạch hơn cho document có cấu trúc phức tạp.
• pdf2image + Tesseract OCR: khi PDF là scan (không có text layer). Convert từng trang thành ảnh rồi OCR.

Với PDF nhỏ đến trung bình (dưới 50 MB, dưới vài trăm trang), đọc toàn bộ vào RAM rồi parse là đủ. Streaming không cần thiết ở đây vì parser cần truy cập ngẫu nhiên vào byte offset của file.

Với file lớn (video, audio dài, large binary), gọi await file.read() một lần nạp toàn bộ vào RAM. Thay vào đó, đọc theo chunk và xử lý hoặc ghi xuống disk từng đợt:

import aiofiles
from fastapi import FastAPI, UploadFile

app = FastAPI()

CHUNK_SIZE = 1024 * 1024  # 1 MB mỗi chunk

@app.post("/upload-large")
async def upload_large(file: UploadFile):
    out_path = f"/tmp/{file.filename}"
    total_bytes = 0

    async with aiofiles.open(out_path, "wb") as out_file:
        while True:
            chunk = await file.read(CHUNK_SIZE)
            if not chunk:
                break
            await out_file.write(chunk)
            total_bytes += len(chunk)

    return {"saved_to": out_path, "total_bytes": total_bytes}

UploadFile.read(size) trả về tối đa size byte. Khi hết dữ liệu, trả về b"". Vòng lặp while True: chunk = ...; if not chunk: break là pattern chuẩn để đọc từng phần.

aiofiles cần cài thêm (pip install aiofiles). Nếu không dùng aiofiles, ghi file đồng bộ trong async def sẽ block event loop — dùng await asyncio.to_thread(write_sync, ...) là alternative.

Khi nào cần streaming?

• File video/audio lớn (vài trăm MB đến GB)
• Audio cho ASR (Automatic Speech Recognition) dài — stream từng đoạn vào model
• Khi bạn cần ghi file xuống S3 mà không muốn buffer toàn bộ trong RAM

Với ảnh JPEG/PNG thông thường (dưới 10 MB) và PDF tài liệu (dưới 50 MB), không cần streaming — await file.read() đủ dùng.

FastAPI không có cơ chế giới hạn kích thước file tích hợp sẵn. Nếu user upload file 10 GB và bạn gọi await file.read(), server sẽ cố đọc 10 GB vào RAM và có thể bị OOM (Out of Memory).

Cách 1: Giới hạn ở reverse proxy (nginx)

# nginx.conf
server {
    client_max_body_size 20M;  # reject request > 20 MB, trả 413 Request Entity Too Large
    ...
}

Đây là cách đơn giản và hiệu quả nhất. Nginx từ chối request trước khi nó đến FastAPI.

Cách 2: Middleware kiểm tra Content-Length

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

MAX_UPLOAD_SIZE = 20 * 1024 * 1024  # 20 MB

@app.middleware("http")
async def limit_upload_size(request: Request, call_next):
    content_length = request.headers.get("content-length")
    if content_length:
        if int(content_length) > MAX_UPLOAD_SIZE:
            return JSONResponse(
                status_code=413,
                content={"detail": f"File too large. Max {MAX_UPLOAD_SIZE // (1024*1024)} MB."},
            )
    return await call_next(request)

Middleware này chỉ kiểm tra header Content-Length. Hầu hết HTTP client gửi header này đúng, nhưng client có thể bỏ qua header hoặc gửi sai. Vì vậy đây là lớp kiểm tra thứ cấp, không thay thế được giới hạn ở reverse proxy.

Để kiểm tra chặt hơn (dừng đọc khi vượt ngưỡng ngay cả khi không có Content-Length), cần đếm bytes khi đọc chunk và raise exception khi tổng vượt giới hạn:

@app.post("/upload-safe")
async def upload_safe(file: UploadFile):
    MAX = 20 * 1024 * 1024  # 20 MB
    total = 0
    chunks = []
    while True:
        chunk = await file.read(1024 * 64)  # 64 KB mỗi chunk
        if not chunk:
            break
        total += len(chunk)
        if total > MAX:
            raise HTTPException(status_code=413, detail="File too large.")
        chunks.append(chunk)
    contents = b"".join(chunks)
    ...

Lưu tạm với tempfile

Khi cần đường dẫn file trên disk (ví dụ thư viện yêu cầu file path thay vì bytes object), dùng tempfile.NamedTemporaryFile:

import tempfile
import os
from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/process")
async def process_file(file: UploadFile):
    contents = await file.read()

    # delete=False để file tồn tại sau khi context manager đóng
    with tempfile.NamedTemporaryFile(
        delete=False,
        suffix=os.path.splitext(file.filename or "")[1],
    ) as tmp:
        tmp.write(contents)
        tmp_path = tmp.name

    try:
        # Truyền đường dẫn cho thư viện bên ngoài
        result = some_library_needing_file_path(tmp_path)
    finally:
        os.unlink(tmp_path)   # xóa file tạm sau khi xong

    return result

Lưu dài hạn — object storage

Với production, lưu file upload lên object storage (S3, MinIO, GCS) thay vì disk local. Lý do: disk local không được chia sẻ giữa nhiều instance, mất dữ liệu khi container restart.

import boto3
from fastapi import FastAPI, UploadFile

app = FastAPI()
s3 = boto3.client("s3")

@app.post("/upload-s3")
async def upload_to_s3(file: UploadFile):
    contents = await file.read()
    bucket = "my-ai-uploads"
    key = f"uploads/{file.filename}"

    s3.put_object(Bucket=bucket, Key=key, Body=contents)
    return {"s3_key": key}

Chi tiết cấu hình IAM, presigned URL, và MinIO self-hosted sẽ được đề cập ở Module 6 (Containerization và Deployment).

curl

# Upload 1 file ảnh
curl -X POST \
  -F "file=@/path/to/image.jpg" \
  http://localhost:8000/classify

# Upload file + form fields
curl -X POST \
  -F "file=@/path/to/image.jpg" \
  -F "model_version=v2" \
  -F "threshold=0.8" \
  http://localhost:8000/classify

# Upload PDF
curl -X POST \
  -F "file=@/path/to/document.pdf" \
  http://localhost:8000/ingest

TestClient (httpx-based) — unit test

from fastapi.testclient import TestClient
import io

# Giả sử app đã định nghĩa ở file main.py
from main import app

client = TestClient(app)

def test_classify_image():
    # Tạo ảnh RGB 1x1 pixel giả
    from PIL import Image
    img = Image.new("RGB", (224, 224), color=(128, 64, 32))
    buf = io.BytesIO()
    img.save(buf, format="JPEG")
    buf.seek(0)

    response = client.post(
        "/classify",
        files={"file": ("test.jpg", buf, "image/jpeg")},
    )
    assert response.status_code == 200
    data = response.json()
    assert "top5" in data
    assert len(data["top5"]) == 5

def test_classify_invalid_type():
    response = client.post(
        "/classify",
        files={"file": ("malware.exe", b"MZ\x90\x00", "image/jpeg")},
    )
    # content_type="image/jpeg" nhưng nội dung là exe —
    # bài này trả 200 vì chỉ check content_type header.
    # Ở production thêm magic byte check để bắt trường hợp này.
    # Test dưới đây kiểm tra trường hợp sai content_type rõ ràng:
    response2 = client.post(
        "/classify",
        files={"file": ("file.pdf", b"%PDF-1.4", "application/pdf")},
    )
    assert response2.status_code == 415

httpx async client

import httpx
import asyncio

async def test_classify_async():
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        with open("image.jpg", "rb") as f:
            response = await client.post(
                "/classify",
                files={"file": ("image.jpg", f, "image/jpeg")},
            )
    print(response.json())

asyncio.run(test_classify_async())

1. Đọc file hai lần mà không seek về 0

Sau khi await file.read() lần đầu, con trỏ đang ở cuối file. Lần gọi tiếp theo trả về b"".

# LỖI: lần 2 trả về b""
contents1 = await file.read()   # đọc toàn bộ, con trỏ ở cuối
contents2 = await file.read()   # b"" — rỗng

# ĐÚNG: seek về 0 trước khi đọc lại
contents1 = await file.read()
await file.seek(0)
contents2 = await file.read()   # OK — cùng dữ liệu

2. Quên đóng file trong finally

Starlette tự cleanup UploadFile sau request, nhưng nếu code tạo nhiều file object hoặc giữ reference lâu, gọi await file.close() trong finally là best practice:

@app.post("/process")
async def process(file: UploadFile):
    try:
        contents = await file.read()
        # ... xử lý ...
    finally:
        await file.close()

3. Validate content_type không đủ — magic byte check

HTTP Content-Type header do client tự gửi. Client có thể đặt tên file malware.exe nhưng khai báo Content-Type: image/jpeg. Endpoint sẽ chấp nhận và cố parse bằng Pillow.

Với production, thêm kiểm tra magic bytes (file signature) bằng python-magic:

import magic   # pip install python-magic

def check_magic_bytes(data: bytes, allowed_mimes: set[str]) -> bool:
    mime = magic.from_buffer(data[:2048], mime=True)
    return mime in allowed_mimes

@app.post("/classify-strict")
async def classify_strict(file: UploadFile):
    contents = await file.read()
    if not check_magic_bytes(contents, {"image/jpeg", "image/png", "image/webp"}):
        raise HTTPException(status_code=415, detail="File content does not match declared type.")
    ...

python-magic đọc vài byte đầu (file signature) để xác định MIME thực sự, bất kể header khai báo gì. Trên Linux/macOS cần libmagic (brew install libmagic hoặc apt install libmagic1).

4. Quên cài python-multipart

FastAPI yêu cầu python-multipart để parse multipart/form-data. Thiếu package này sẽ nhận 422 Unprocessable Entity với message "Field required" cho mọi upload endpoint dù code khai báo đúng.

5. Dùng sync handler với UploadFile

UploadFile.read(), .seek(), .close() đều là coroutine, chỉ dùng được trong async def. Nếu dùng def (sync), bạn không thể await — kết quả của file.read() sẽ là coroutine object chứ không phải bytes. Luôn khai báo upload endpoint là async def.

• ✅ Mặc định dùng UploadFile — có SpooledTemporaryFile, có .filename và .content_type, không load hết vào RAM khi file lớn
• ✅ await file.read() là coroutine — endpoint phải là async def
• ✅ Validate content_type để trả 415 sớm; thêm magic byte check cho production
• ✅ Batch upload: files: list[UploadFile] — stack tensors thành batch rồi forward 1 lần
• ✅ Form field kèm file dùng Form(...), không dùng Body(...) trong cùng multipart request
• ✅ PDF parsing: pypdf 4.x cho text, pdfplumber cho table, unstructured cho layout phức tạp
• ✅ Streaming chunk: while chunk := await file.read(CHUNK_SIZE) cho file lớn
• ✅ Giới hạn kích thước: nginx client_max_body_size hoặc middleware đếm bytes
• ✅ Đọc file lần 2 phải await file.seek(0) trước; gọi await file.close() trong finally

Bài 8: Streaming response cho LLM API — trả kết quả từng token khi model đang sinh, tránh timeout với request có latency cao.