Bài 3: Pydantic models — validate request và response

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

• Nắm được các thay đổi breaking của Pydantic v2 so với v1.
• Định nghĩa BaseModel cho request body và gắn vào endpoint FastAPI.
• Dùng response_model để kiểm soát field trả về.
• Thêm constraint vào field bằng Field(...).
• Viết validator tuỳ chỉnh với @field_validator và @model_validator.
• Xây nested model cho batch request.
• Hiểu khi nào dùng Optional, Union, Literal.
• Tránh các pitfall phổ biến: mutable default, model_dump mode, trả về dict thay vì model instance.

FastAPI 0.100+ (tháng 7/2023) hỗ trợ Pydantic v2 song song. FastAPI 0.110+ yêu cầu Pydantic v2 — v1 không còn được hỗ trợ. Nếu dự án đang dùng v1, migration là bắt buộc.

Thay đổi lớn nhất của v2: core validation được viết lại bằng Rust (pydantic-core), nhanh hơn v1 từ 5× đến 50× tuỳ workload. Ngoài tốc độ, v2 thay đổi một số API bề mặt (breaking changes):

# Pydantic v1 — KHÔNG dùng với FastAPI 0.110+
class Config:
    orm_mode = True

obj.dict()          # serialize sang dict
MyModel.parse_obj(data)  # validate từ dict
@validator("field")       # validator cũ
def check_field(cls, v): ...

# Pydantic v2 — syntax mới
from pydantic import ConfigDict

model_config = ConfigDict(from_attributes=True)  # thay orm_mode

obj.model_dump()         # thay .dict()
MyModel.model_validate(data)  # thay parse_obj
@field_validator("field")     # thay @validator
def check_field(cls, v): ...

Tóm tắt các thay đổi cần nhớ:

• class Config → model_config = ConfigDict(...)
• .dict() → .model_dump()
• .json() → .model_dump_json()
• .parse_obj() → .model_validate()
• .parse_raw() → .model_validate_json()
• @validator → @field_validator
• @root_validator → @model_validator
• orm_mode = True → ConfigDict(from_attributes=True)

Cài đặt đúng version:

pip install "fastapi[standard]" "pydantic>=2.0"

Kiểm tra:

import pydantic
print(pydantic.VERSION)  # phải là 2.x.x

Khi endpoint cần nhận JSON body (POST, PUT, PATCH), khai báo tham số kiểu BaseModel. FastAPI tự parse body, validate kiểu, và raise 422 Unprocessable Entity nếu dữ liệu không hợp lệ — không cần viết code kiểm tra thủ công.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class PredictRequest(BaseModel):
    features: list[float]
    model_version: str = "v1"  # có default — field tuỳ chọn

@app.post("/predict")
async def predict(req: PredictRequest):
    # req đã được validate: req.features là list[float], req.model_version là str
    return {"received_features": len(req.features), "model": req.model_version}

Thử với httpx hoặc curl:

# Request hợp lệ
curl -X POST http://127.0.0.1:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [0.1, 0.2, 0.3], "model_version": "v2"}'

# Thiếu field bắt buộc → 422
curl -X POST http://127.0.0.1:8000/predict \
  -H "Content-Type: application/json" \
  -d '{}'

# features không phải list[float] → 422
curl -X POST http://127.0.0.1:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": "bad_input"}'

Response lỗi 422 từ FastAPI có cấu trúc cố định:

{
  "detail": [
    {
      "type": "missing",
      "loc": ["body", "features"],
      "msg": "Field required",
      "input": {}
    }
  ]
}

Field không có default (như features: list[float]) là bắt buộc. Field có default (như model_version: str = "v1") là tuỳ chọn — nếu client không gửi, FastAPI dùng giá trị default.

Pydantic v2 cố gắng coerce (ép kiểu) khi có thể trong lax mode — ví dụ "1.5" có thể được chấp nhận cho field float. Để tắt coercion, dùng model_config = ConfigDict(strict=True).

Khai báo response_model trong decorator của endpoint để:

• FastAPI lọc bỏ field không có trong schema trước khi trả về client — tránh leak field nội bộ.
• FastAPI validate response trước khi gửi — phát hiện bug ở tầng response sớm.
• OpenAPI docs tự sinh schema cho response, client/frontend đọc được cấu trúc mà không cần hỏi.

from pydantic import BaseModel

class PredictRequest(BaseModel):
    features: list[float]
    model_version: str = "v1"

class PredictResponse(BaseModel):
    prediction: float
    confidence: float
    model_version: str
    # KHÔNG có: internal_model_id, raw_logits — các field này sẽ bị lọc

@app.post("/predict", response_model=PredictResponse)
async def predict(req: PredictRequest):
    # giả lập inference
    raw_result = {
        "prediction": 0.87,
        "confidence": 0.92,
        "model_version": req.model_version,
        "internal_model_id": "xgb-prod-20240501",  # field nội bộ
        "raw_logits": [0.13, 0.87],                 # field nội bộ
    }
    # Cách 1: trả dict → FastAPI tự validate và lọc theo response_model
    return raw_result

    # Cách 2: explicit — rõ ràng hơn, khuyến nghị
    # return PredictResponse(
    #     prediction=raw_result["prediction"],
    #     confidence=raw_result["confidence"],
    #     model_version=raw_result["model_version"],
    # )

Khi dùng Cách 1 (trả dict), FastAPI validate qua response_model. Khi dùng Cách 2 (trả model instance), validation xảy ra trong constructor PredictResponse(...). Cách 2 thường rõ ràng hơn trong code lớn.

Trường hợp muốn bỏ response validation (hiếm, chỉ khi cần tốc độ tối đa và data đã được validate trước):

@app.post("/predict", response_model=PredictResponse, response_model_exclude_unset=True)
async def predict(req: PredictRequest):
    ...

response_model_exclude_unset=True — chỉ serialize field nào được set giá trị tường minh, bỏ qua field có default không được set. Hữu ích khi response có nhiều field tuỳ chọn và muốn JSON nhỏ gọn.

Field từ pydantic cho phép thêm constraint và metadata vào từng field:

from pydantic import BaseModel, Field

class PredictRequest(BaseModel):
    features: list[float] = Field(
        ...,                     # ... nghĩa là bắt buộc (required)
        min_length=1,            # list phải có ít nhất 1 phần tử
        max_length=1000,         # không quá 1000 phần tử
        description="Vector đặc trưng đầu vào, các giá trị float",
    )
    model_version: str = Field(
        default="v1",
        min_length=1,
        max_length=10,
        description="Phiên bản model, ví dụ 'v1' hoặc 'v2'",
    )

class ClassificationResponse(BaseModel):
    label: str = Field(..., description="Nhãn phân loại")
    probability: float = Field(
        ...,
        ge=0.0,   # greater or equal — xác suất không âm
        le=1.0,   # less or equal — xác suất tối đa 1
        description="Xác suất của nhãn, từ 0.0 đến 1.0",
    )
    latency_ms: float = Field(
        ...,
        gt=0.0,   # strictly greater than 0
        description="Thời gian inference tính bằng millisecond",
    )

Các constraint số phổ biến:

• gt — greater than (lớn hơn, không bao gồm)
• ge — greater or equal (lớn hơn hoặc bằng)
• lt — less than
• le — less or equal
• multiple_of — phải là bội số

Constraint chuỗi:

• min_length, max_length
• pattern — regex pattern (Python re syntax)

Constraint collection (list, set):

• min_length, max_length — số phần tử

Field description xuất hiện trong /docs (Swagger UI) và schema JSON được sinh ra — giúp client hiểu API mà không cần đọc code.

# Ví dụ field dùng pattern
class TextRequest(BaseModel):
    text: str = Field(
        ...,
        min_length=1,
        max_length=512,
        description="Text cần phân tích",
    )
    language: str = Field(
        default="vi",
        pattern=r"^[a-z]{2}$",  # ISO 639-1: 2 ký tự thường
        description="Mã ngôn ngữ ISO 639-1 (vd: 'vi', 'en', 'ja')",
    )

Khi constraint dựng sẵn không đủ, dùng validator tuỳ chỉnh. Pydantic v2 dùng @field_validator cho validator 1 field, @model_validator cho validator cần đọc nhiều field cùng lúc.

@field_validator — validate từng field

from pydantic import BaseModel, Field, field_validator

class PredictRequest(BaseModel):
    features: list[float] = Field(..., min_length=1)
    model_version: str = Field(default="v1")

    @field_validator("model_version")
    @classmethod
    def version_must_be_known(cls, v: str) -> str:
        allowed = {"v1", "v2", "v3"}
        if v not in allowed:
            raise ValueError(f"model_version phải là một trong {allowed}, nhận được: {v!r}")
        return v  # phải return giá trị

    @field_validator("features")
    @classmethod
    def features_must_be_finite(cls, v: list[float]) -> list[float]:
        import math
        for i, x in enumerate(v):
            if not math.isfinite(x):
                raise ValueError(f"features[{i}] = {x} không hợp lệ (NaN hoặc Inf)")
        return v

Lưu ý cú pháp v2:

• Phải thêm @classmethod ngay sau @field_validator.
• Phải return giá trị. Nếu quên return, field sẽ là None.
• Validator nhận giá trị đã được coerce (ví dụ "1.5" đã thành 1.5). Dùng mode="before" để chạy trước coercion.

mode='before' — chạy trước type coercion

from pydantic import field_validator

class PredictRequest(BaseModel):
    features: list[float]

    @field_validator("features", mode="before")
    @classmethod
    def parse_features_string(cls, v):
        # Cho phép client gửi features dưới dạng string JSON: "[0.1,0.2]"
        if isinstance(v, str):
            import json
            try:
                parsed = json.loads(v)
                if not isinstance(parsed, list):
                    raise ValueError("features string phải là JSON array")
                return parsed
            except json.JSONDecodeError as e:
                raise ValueError(f"features không phải JSON hợp lệ: {e}")
        return v  # đã là list, trả nguyên

@model_validator — validate tương tác giữa nhiều field

from pydantic import BaseModel, Field, model_validator
from typing import Self

# Giả sử: v1 cần đúng 10 features, v2 cần đúng 20 features
EXPECTED_FEATURES = {"v1": 10, "v2": 20}

class PredictRequest(BaseModel):
    features: list[float] = Field(..., min_length=1)
    model_version: str = Field(default="v1")

    @model_validator(mode="after")
    def check_feature_count(self) -> Self:
        expected = EXPECTED_FEATURES.get(self.model_version)
        if expected is not None and len(self.features) != expected:
            raise ValueError(
                f"model_version='{self.model_version}' cần {expected} features, "
                f"nhận được {len(self.features)}"
            )
        return self  # phải return self

mode="after" (mặc định cho model_validator) chạy sau khi tất cả field đã được parse và validate — dùng self.field_name bình thường. mode="before" nhận raw dict chưa được parse.

Khi validator raise ValueError, FastAPI bắt và trả về lỗi 422 với cấu trúc chuẩn. Không cần raise HTTPException bên trong validator.

Model có thể chứa model khác làm field. Validation đệ quy tự động — Pydantic validate từng cấp theo thứ tự từ trong ra ngoài.

from pydantic import BaseModel, Field

class TextChunk(BaseModel):
    chunk_id: str = Field(..., min_length=1, description="ID duy nhất của chunk")
    text: str = Field(..., min_length=1, max_length=2048, description="Nội dung text")
    metadata: dict[str, str] = Field(default_factory=dict, description="Key-value metadata tuỳ ý")

class ChunkBatch(BaseModel):
    chunks: list[TextChunk] = Field(
        ...,
        min_length=1,
        max_length=64,
        description="Danh sách chunk cần encode, tối đa 64 chunk mỗi request",
    )
    model_name: str = Field(
        default="text-embedding-3-small",
        description="Tên embedding model",
    )

@app.post("/embed/batch")
async def embed_batch(batch: ChunkBatch):
    # batch.chunks là list[TextChunk], mỗi phần tử đã validated
    return {"count": len(batch.chunks), "model": batch.model_name}

JSON request tương ứng:

{
  "chunks": [
    {"chunk_id": "c1", "text": "Pydantic là thư viện validation Python"},
    {"chunk_id": "c2", "text": "FastAPI dùng Pydantic để parse request"}
  ],
  "model_name": "text-embedding-3-small"
}

Nếu bất kỳ chunk nào sai (vd text rỗng), lỗi 422 trả về đúng path: ["body", "chunks", 1, "text"]. Client biết chính xác phần tử nào lỗi.

Nested model cũng dùng cho response:

class EmbedResult(BaseModel):
    chunk_id: str
    embedding: list[float]
    token_count: int

class EmbedBatchResponse(BaseModel):
    results: list[EmbedResult]
    model_name: str
    total_tokens: int

@app.post("/embed/batch", response_model=EmbedBatchResponse)
async def embed_batch(batch: ChunkBatch):
    ...

Optional

Optional[X] là shorthand cho Union[X, None]. Trong Python 3.10+ có thể viết X | None.

from typing import Optional
from pydantic import BaseModel, Field

class PredictRequest(BaseModel):
    features: list[float]
    model_version: str = "v1"

    # Field tuỳ chọn, mặc định None nếu client không gửi
    request_id: Optional[str] = None
    # Python 3.10+ syntax tương đương:
    # request_id: str | None = None

Chú ý: Optional[str] mà không đặt default thì field vẫn là bắt buộc (chỉ có thể là str hoặc None). Phải đặt = None để field thực sự tuỳ chọn.

# field bắt buộc — phải gửi, nhưng có thể gửi null
required_nullable: Optional[str]   # client phải gửi, nhưng được gửi null

# field tuỳ chọn — không gửi được, mặc định None
optional_nullable: Optional[str] = None

Literal — enum nhẹ

Literal ràng buộc field chỉ nhận một tập giá trị cố định — tương đương enum nhưng nhẹ hơn, schema OpenAPI rõ hơn.

from typing import Literal

class TextRequest(BaseModel):
    text: str
    task: Literal["classify", "summarize", "translate"] = "classify"
    model_size: Literal["small", "base", "large"] = "base"

OpenAPI docs sẽ hiển thị dropdown với các giá trị hợp lệ. Nếu client gửi giá trị ngoài tập, nhận 422 ngay.

Union — đa kiểu

from typing import Union

class FlexRequest(BaseModel):
    # input có thể là string hoặc list of string
    input: Union[str, list[str]]

Pydantic v2 strict mode xử lý Union khác v1: trong strict mode, không coerce kiểu — phải gửi đúng kiểu khai báo. Trong lax mode (mặc định), Pydantic thử từng kiểu trong Union theo thứ tự và dùng kiểu đầu tiên khớp. Thứ tự Union ảnh hưởng đến kết quả:

# Cẩn thận với thứ tự Union:
class A(BaseModel):
    v: Union[int, str]  # "123" sẽ được coerce thành int 123 vì int đứng trước

class B(BaseModel):
    v: Union[str, int]  # "123" giữ nguyên là str "123" vì str đứng trước

Python 3.10+ syntax:

class FlexRequest(BaseModel):
    input: str | list[str]  # tương đương Union[str, list[str]]

Pitfall 1: trả dict thay vì model instance khi không dùng response_model

Khi không khai báo response_model, trả về dict bỏ qua hoàn toàn Pydantic validation. Nếu logic có bug sinh ra field sai kiểu, client nhận data bẩn mà không báo lỗi.

# CÓ VẤN ĐỀ — không có response_model
@app.post("/predict")
async def predict(req: PredictRequest):
    # nếu code sai, raw_result có thể có field sai kiểu — client nhận vô tư
    return {"prediction": "accidentally_a_string"}

# TỐT HƠN — khai báo rõ response_model
@app.post("/predict", response_model=PredictResponse)
async def predict(req: PredictRequest):
    return PredictResponse(prediction=0.87, confidence=0.92, model_version="v1")

Pitfall 2: mutable default value

Pydantic v2 raise PydanticUserError nếu dùng list, dict hoặc object khác làm default trực tiếp. Lý do: Python tái sử dụng cùng object cho tất cả instance, dẫn đến chia sẻ state ngoài ý muốn.

from pydantic import BaseModel, Field

# SAI — Pydantic v2 raise lỗi ngay khi định nghĩa class
class BadModel(BaseModel):
    tags: list[str] = []           # lỗi PydanticUserError
    metadata: dict = {}            # lỗi PydanticUserError

# ĐÚNG — dùng Field(default_factory=...)
class GoodModel(BaseModel):
    tags: list[str] = Field(default_factory=list)
    metadata: dict = Field(default_factory=dict)
    scores: list[float] = Field(default_factory=lambda: [0.0, 0.0])

Pitfall 3: model_dump() vs model_dump(mode="json")

.model_dump() trả Python dict giữ nguyên kiểu Python (datetime, UUID giữ nguyên object). .model_dump(mode="json") serialize tất cả thành kiểu JSON-compatible (datetime thành ISO string, UUID thành string).

from datetime import datetime
from uuid import UUID, uuid4
from pydantic import BaseModel

class Event(BaseModel):
    event_id: UUID
    created_at: datetime

e = Event(event_id=uuid4(), created_at=datetime.now())

d1 = e.model_dump()
# {"event_id": UUID("..."), "created_at": datetime(...)}  — kiểu Python

d2 = e.model_dump(mode="json")
# {"event_id": "550e8400-...", "created_at": "2026-05-27T..."}  — JSON-safe

# Khi muốn log hoặc truyền sang JSON serializer bên ngoài, dùng mode="json"
import json
json.dumps(d2)   # OK
# json.dumps(d1) → TypeError: Object of type UUID is not JSON serializable

Pitfall 4: nhầm @validator (v1) sang @field_validator (v2)

Dùng @validator (v1) với Pydantic v2 vẫn chạy nhưng bị deprecation warning và có hành vi khác. Đặc biệt, @validator cũ không cần @classmethod riêng, còn @field_validator v2 bắt buộc có.

# v1 — KHÔNG dùng với Pydantic v2
from pydantic import validator

class OldModel(BaseModel):
    name: str

    @validator("name")
    def name_not_empty(cls, v):
        ...

# v2 — đúng
from pydantic import field_validator

class NewModel(BaseModel):
    name: str

    @field_validator("name")
    @classmethod           # bắt buộc
    def name_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("name không được rỗng hoặc chỉ có khoảng trắng")
        return v

Pitfall 5: quên return trong validator

# SAI — quên return → field sẽ là None
@field_validator("features")
@classmethod
def check_features(cls, v: list[float]) -> list[float]:
    if len(v) == 0:
        raise ValueError("features không được rỗng")
    # quên return v → validator trả None → field thành None

# ĐÚNG
@field_validator("features")
@classmethod
def check_features(cls, v: list[float]) -> list[float]:
    if len(v) == 0:
        raise ValueError("features không được rỗng")
    return v  # bắt buộc

1. Request + response đơn giản: Viết endpoint POST /classify nhận {"text": "...", "top_k": 3}. text bắt buộc, max_length=1000. top_k là int, mặc định 1, phải trong khoảng 1–10. Response trả {"labels": [...], "scores": [...]} với response_model. Dùng giá trị giả cho labels/scores. Kiểm tra: gửi request thiếu text, gửi top_k=0, gửi top_k=11 — quan sát response 422.
2. @field_validator: Thêm validator vào endpoint ở bài 1 để kiểm tra text sau khi strip() không được rỗng (Field(min_length=1) không bắt được chuỗi toàn khoảng trắng). Gửi {"text": " "} — kỳ vọng nhận 422.
3. @model_validator: Tạo model BatchRequest với field texts: list[str] và max_batch_size: int = 8. Thêm @model_validator kiểm tra len(texts) <= max_batch_size — nếu vượt, raise ValueError mô tả rõ.
4. Nested model + response lọc field: Định nghĩa DocumentChunk(BaseModel) với doc_id, text, page_number: Optional[int] = None. Định nghĩa IngestRequest với chunks: list[DocumentChunk]. Response trả chỉ {"ingested_count": N}, không lộ thông tin chunk. Kiểm tra: gửi chunk thiếu text, gửi page_number=-1 (thêm constraint ge=1).
5. Pitfall check: (a) Định nghĩa model có tags: list[str] = [] — quan sát lỗi Pydantic. Sửa bằng default_factory. (b) Viết model có created_at: datetime, tạo instance, in .model_dump() và .model_dump(mode="json") — quan sát kiểu của created_at trong từng kết quả.