Python hiện đại 5 — Type hints & Pydantic v2: dữ liệu đáng tin

Trong Tổng quan Python hiện đại cho Data, chúng ta điểm qua bộ công cụ giúp code dữ liệu vừa nhanh vừa an toàn. Bài này đi vào hai trụ cột của phần "an toàn": type hints (chú thích kiểu) và Pydantic v2. Python vốn là ngôn ngữ động (dynamic typing) — biến nào cũng có thể chứa bất cứ thứ gì. Điều đó dễ viết nhưng cực kỳ nguy hiểm với code dữ liệu, nơi một trường amount là chuỗi "1,000" thay vì số có thể lặng lẽ làm sai cả một báo cáo cuối tháng. Type hints và Pydantic là hai cách trám lỗ hổng đó: một cái ở tầng công cụ tĩnh, một cái ở tầng chạy thật (runtime).

Type hints: vì sao quan trọng với code dữ liệu

Type hint là cú pháp cho phép bạn ghi rõ kiểu mong đợi của biến, tham số, giá trị trả về:

def tinh_phi(so_tien: float, ty_le: float) -> float:
    return so_tien * ty_le

Quan trọng: Python không tự ép hay kiểm type hint khi chạy — chúng chỉ là chú thích. Giá trị nằm ở ba chỗ khác:

• Bắt lỗi sớm: công cụ kiểm tĩnh (mypy, pyright) đọc hint và cảnh báo trước khi chạy nếu bạn truyền str vào chỗ cần float, hay quên xử lý None. Với pipeline chạy hàng đêm, phát hiện lỗi lúc code còn trên máy dev rẻ hơn rất nhiều so với lúc job đổ vỡ lúc 2 giờ sáng.
• Tự tài liệu hoá: chữ ký hàm nói rõ nó nhận gì, trả gì. Người sau (kể cả chính bạn sáu tháng sau) không phải đọc hết thân hàm để đoán.
• IDE thông minh hơn: autocomplete, refactor, "go to definition" đều chính xác hơn khi công cụ biết kiểu.

Cú pháp hiện đại

Từ Python 3.10+ cú pháp gọn hơn hẳn thời phải from typing import List, Optional:

# Danh sách, dict dùng built-in generic (3.9+)
so_du: list[float]
ho_so: dict[str, int]

# Union và optional dùng dấu | (3.10+)
ma_chi_nhanh: str | None          # thay cho Optional[str]
gia_tri: int | str

# Literal: chỉ nhận đúng vài giá trị cố định
from typing import Literal
loai_gd: Literal["credit", "debit"]

# TypedDict: mô tả cấu trúc một dict (bản ghi dạng dict)
from typing import TypedDict
class BanGhi(TypedDict):
    account_no: str
    amount: float

# Protocol: "structural typing" — hợp lệ nếu có đủ method, không cần kế thừa
from typing import Protocol
class CoDoc(Protocol):
    def read(self) -> bytes: ...

• Literal cực hữu ích cho các trường có tập giá trị hữu hạn (loại giao dịch, mã tiền tệ), mypy sẽ chặn nếu bạn lỡ viết "debitt".
• TypedDict mô tả kiểu cho dữ liệu dạng dict mà không cần đổi sang class — hợp khi dữ liệu đến từ JSON/API.
• Protocol cho phép "duck typing có kiểm soát": bất kỳ đối tượng nào có method read() đều hợp lệ, không bắt buộc kế thừa lớp cha.
• Generics (TypeVar) giúp viết hàm/lớp dùng lại cho nhiều kiểu mà vẫn giữ được thông tin kiểu:

from typing import TypeVar
T = TypeVar("T")
def phan_tu_dau(xs: list[T]) -> T:
    return xs[0]        # trả về đúng kiểu của phần tử

Kiểm tĩnh với mypy / pyright

Type hint chỉ phát huy khi có công cụ đọc chúng. Hai lựa chọn phổ biến: mypy (chuẩn mực, dự án cộng đồng) và pyright (nhanh, đứng sau VS Code/Pylance).

mypy src/            # quét toàn bộ, báo lỗi type không khớp
pyright src/

Nên bật kiểm tĩnh trong CI để mỗi commit đều được soi. Đây là một mắt xích của chất lượng code Python, sẽ nói kỹ hơn trong Kiểm thử & chất lượng. Lưu ý: kiểm tĩnh không thay được validate runtime — nó không biết dữ liệu thực từ file/API sẽ ra sao. Đó là lúc Pydantic vào cuộc.

Dataclass vs Pydantic

Cả hai đều giúp định nghĩa "cấu trúc dữ liệu có tên trường". Khác biệt cốt lõi: dataclass chỉ tổ chức, Pydantic còn kiểm và ép kiểu.

@dataclass (chuẩn trong thư viện dataclasses) sinh sẵn __init__, __repr__, __eq__ từ khai báo trường. Nó nhẹ, không phụ thuộc gói ngoài, nhưng không validate:

from dataclasses import dataclass

@dataclass
class GiaoDich:
    account_no: str
    amount: float

gd = GiaoDich(account_no=123, amount="rất nhiều")  # KHÔNG báo lỗi!

Dataclass nhận tuốt: account_no thành số, amount thành chuỗi vô nghĩa — không ai kêu ca cho tới khi code dùng chúng vỡ ở đâu đó xa. Dataclass hợp khi dữ liệu đã sạch, hoặc bạn chỉ cần một "túi trường" nội bộ, hiệu năng tối đa và không muốn thêm dependency.

Pydantic thì ngược lại: nó là "người gác cổng". Khi bạn tạo model, Pydantic kiểm và ép kiểu ngay lúc chạy, và ném lỗi rõ ràng nếu dữ liệu không hợp lệ.

Pydantic v2: gác cổng dữ liệu

Pydantic v2 (ra 2023) viết lại phần lõi (pydantic-core) bằng Rust, nhanh hơn v1 nhiều lần cho cùng khối lượng validate — đây là khác biệt lớn nhất so với v1 thuần Python. Model được khai báo bằng cách kế thừa BaseModel và dùng chính type hint làm quy tắc:

from pydantic import BaseModel, Field, EmailStr

class KhachHang(BaseModel):
    id: int
    full_name: str = Field(min_length=1, max_length=100)
    email: EmailStr
    tuoi: int = Field(ge=18, le=120)   # 18 ≤ tuoi ≤ 120

• Field: gắn ràng buộc (độ dài, khoảng giá trị, giá trị mặc định, mô tả) vào từng trường.
• Kiểu ràng buộc dựng sẵn: EmailStr (kiểm định dạng email), conint(gt=0), constr(pattern=...), PositiveInt, AnyUrl... để diễn đạt ràng buộc phổ biến gọn gàng.

Validator: luật nghiệp vụ

Ràng buộc trường đơn giản thì dùng Field. Luật phức tạp hơn dùng validator:

from pydantic import field_validator, model_validator

class ChuyenKhoan(BaseModel):
    tk_gui: str
    tk_nhan: str
    so_tien: float

    @field_validator("so_tien")
    @classmethod
    def duong(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("số tiền phải > 0")
        return v

    @model_validator(mode="after")
    def khac_tai_khoan(self):
        if self.tk_gui == self.tk_nhan:
            raise ValueError("tài khoản gửi và nhận không được trùng")
        return self

• field_validator: kiểm/biến đổi một trường.
• model_validator(mode="after"): kiểm toàn model sau khi các trường đã dựng — hợp cho luật liên quan nhiều trường.

Parse dữ liệu bẩn → model sạch

Đây là sức mạnh thực sự. Pydantic không chỉ kiểm — nó parse: nhận dữ liệu lộn xộn (chuỗi từ CSV, JSON từ API) và ép về kiểu đúng.

data = {"id": "42", "full_name": "Nguyen Van A",
        "email": "[email protected]", "tuoi": "35"}
kh = KhachHang.model_validate(data)
# kh.id == 42 (int), kh.tuoi == 35 (int) — chuỗi đã được ép

kh.model_dump()          # -> dict Python sạch
kh.model_dump_json()     # -> chuỗi JSON
KhachHang.model_json_schema()  # -> JSON Schema mô tả model

• model_validate(dict) / model_validate_json(str): dựng model từ dữ liệu, validate luôn.
• model_dump() / model_dump_json(): xuất ngược ra dict / JSON để lưu hoặc gửi đi.
• model_json_schema(): sinh JSON Schema — dùng làm tài liệu API hoặc "hợp đồng dữ liệu" chia sẻ với đối tác.

pydantic-settings: cấu hình từ env

Gói riêng pydantic-settings đọc cấu hình job từ biến môi trường / file .env, validate luôn kiểu — thay cho việc rải rác os.getenv(...) trả về chuỗi không kiểm:

from pydantic_settings import BaseSettings

class CauHinh(BaseSettings):
    db_url: str
    batch_size: int = 1000
    doc_len_env: bool = True   # tự đọc DB_URL, BATCH_SIZE từ env

cfg = CauHinh()   # thiếu db_url -> báo lỗi ngay khi khởi động

Lợi ích: job fail nhanh lúc khởi động nếu cấu hình sai/thiếu, thay vì chạy nửa chừng mới vỡ.

Ứng dụng trong công việc dữ liệu

Pydantic tỏa sáng ở mọi ranh giới — nơi dữ liệu từ ngoài đi vào code của bạn:

• Đầu vào pipeline: validate từng bản ghi giao dịch/khách hàng trước khi nạp. Bản ghi thiếu trường, sai kiểu bị chặn ngay, không "ngấm" vào kho.
• Ranh giới API: FastAPI dùng chính Pydantic để validate request/response — dữ liệu client gửi lên được kiểm trước khi chạm logic.
• Cấu hình job: như phần pydantic-settings ở trên.
• "Data contract" ở tầng code: model Pydantic là bản mô tả có thể chạy được của hình dạng dữ liệu bạn cam kết. Nó bổ trợ cho kiểm soát chất lượng dữ liệu ở tầng nghiệp vụ/kho (xem Chất lượng dữ liệu): governance định nghĩa "quy tắc", Pydantic thực thi một phần quy tắc đó ngay tại điểm nhập.

Pydantic phối hợp tốt với Polars/DuckDB: validate rồi mới xử lý. Mẫu điển hình — duyệt dữ liệu thô, Pydantic tách bản ghi hợp lệ khỏi bản ghi lỗi, chỉ phần sạch mới đưa vào Polars DataFrame để tính toán tốc độ cao. Bạn được cả hai: an toàn ở cổng vào và hiệu năng ở khâu tính.

Luồng validate

So với validate bằng tay

Không có Pydantic, người ta viết hàng loạt if: kiểm None, thử float(x) trong try/except, so khớp regex... Cách này dài, dễ bỏ sót, khó bảo trì, và thông báo lỗi mỗi người viết một kiểu. Pydantic gom toàn bộ vào khai báo, gộp mọi lỗi của một bản ghi vào một ValidationError có cấu trúc (đường dẫn trường + thông điệp), rất tiện để log và trả về.

Cạm bẫy cần biết

• Coercion (ép kiểu ngầm): mặc định Pydantic khá "khoan dung" — chuỗi "35" thành int 35, "true" thành bool. Tiện khi parse dữ liệu bẩn, nhưng có thể che giấu dữ liệu lệch chuẩn. Nếu muốn chặt, bật strict mode (model_config = ConfigDict(strict=True) hoặc kiểu StrictInt) để chỉ nhận đúng kiểu, không ép.
• Hiệu năng: validate không miễn phí. Với hàng chục triệu dòng, đừng dựng một model Pydantic cho từng dòng trong vòng lặp Python nếu có thể validate theo lô bằng công cụ vector hoá. Pydantic hợp nhất ở ranh giới (nhập/xuất, API, cấu hình), không phải thay thế phép toán trên DataFrame.
• Nhầm với type hint tĩnh: Pydantic kiểm lúc chạy, mypy kiểm lúc dịch — hai lớp bổ sung nhau, không thay thế.

Ví dụ: model giao dịch ngân hàng (minh hoạ)

Ghép các mảnh lại thành một model cho bản ghi giao dịch như thường gặp trong file đối tác. (Đây là code Python minh hoạ, không phải SQL sandbox.)

from datetime import datetime
from typing import Literal
from pydantic import BaseModel, Field, field_validator, ValidationError

class GiaoDich(BaseModel):
    account_no: str = Field(pattern=r"^\d{8,16}$")   # 8–16 chữ số
    amount: float = Field(gt=0)                       # > 0
    currency: Literal["VND", "USD", "EUR"]
    kind: Literal["credit", "debit"]
    created_at: datetime

    @field_validator("account_no")
    @classmethod
    def khong_toan_so_khong(cls, v: str) -> str:
        if set(v) == {"0"}:
            raise ValueError("số tài khoản không hợp lệ")
        return v

# --- Parse & bắt lỗi ---
tho = [
    {"account_no": "1234567890", "amount": "1500000",
     "currency": "VND", "kind": "credit", "created_at": "2026-07-01T09:30:00"},
    {"account_no": "12", "amount": "-5", "currency": "VND",
     "kind": "credit", "created_at": "2026-07-01T10:00:00"},
]

hop_le, loi = [], []
for i, row in enumerate(tho):
    try:
        hop_le.append(GiaoDich.model_validate(row))
    except ValidationError as e:
        loi.append((i, e.errors()))   # danh sách lỗi có cấu trúc

Bản ghi đầu: amount chuỗi "1500000" được ép về float, created_at chuỗi ISO được parse thành datetime — hợp lệ. Bản ghi thứ hai vi phạm hai luật cùng lúc (account_no chỉ 2 chữ số, amount âm); Pydantic gom cả hai vào một ValidationError, e.errors() trả danh sách chỉ rõ trường nào sai và vì sao. Nhờ vậy ta tách riêng dòng lỗi thay vì để cả lô đổ vỡ.

Use case thực tế

Bối cảnh NCB — kiểm dữ liệu file đối tác trước khi nạp kho. Hằng ngày, đội dữ liệu nhận file CSV/JSON từ các đối tác (ví, cổng thanh toán, tổ chức liên kết) chứa bản ghi giao dịch để đối soát. Trước đây, file được nạp thẳng vào staging rồi mới phát hiện lỗi ở khâu báo cáo — có tháng phải chạy lại cả pipeline vì vài nghìn dòng sai định dạng ngày hoặc thiếu mã tiền tệ.

Đội xây một cổng validate bằng Pydantic đặt ngay sau bước tải file:

1. Định nghĩa model GiaoDichDoiTac mô tả chính xác hợp đồng dữ liệu: các trường bắt buộc, account_no khớp mẫu, amount > 0, currency ∈ {VND, USD, EUR}, created_at là timestamp hợp lệ.
2. Với mỗi dòng, gọi model_validate. Dòng hợp lệ đi tiếp; dòng lỗi bị tách sang file rejects kèm errors() (số dòng + trường + lý do).
3. Phần hợp lệ được đổ vào Polars để chuẩn hoá và nạp kho; phần lỗi được tổng hợp thành báo cáo gửi lại đối tác.

Số liệu ước lượng (minh hoạ, tuỳ chất lượng nguồn): trên lô ~500.000 bản ghi/ngày, cổng bắt được khoảng 0,5–2% dòng lỗi (sai định dạng ngày, thiếu trường, số tiền ≤ 0, mã tiền tệ lạ) mà trước đây lọt vào kho. Thời gian phát hiện lỗi rút từ "cuối chu kỳ báo cáo" xuống ngay lúc nạp, và mỗi lỗi có địa chỉ chính xác (dòng nào, trường nào) nên đối tác sửa nhanh. Quan trọng nhất: kho chỉ chứa dữ liệu đã qua cổng, nên các bước hạ nguồn không còn phải phòng thủ chống dữ liệu rác.

Ghi nhớ

• Type hints không kiểm lúc chạy; giá trị của chúng là bắt lỗi sớm qua mypy/pyright, tự tài liệu hoá và IDE thông minh. Dùng cú pháp hiện đại: list[int], X | None, Literal, TypedDict, Protocol, generics.
• dataclass: nhẹ, chuẩn Python, chỉ tổ chức dữ liệu, không validate. Pydantic v2: validate + parse + serialize lúc chạy, lõi viết bằng Rust nên nhanh.
• Pydantic v2 cốt lõi: BaseModel, Field (ràng buộc), field_validator/model_validator (luật nghiệp vụ), kiểu ràng buộc (EmailStr, conint...), model_validate/model_dump, JSON Schema; pydantic-settings cho cấu hình từ env (fail nhanh khi khởi động).
• Đặt Pydantic ở ranh giới dữ liệu: đầu vào pipeline, API, cấu hình. Đây là "data contract" chạy được, bổ trợ cho chất lượng dữ liệu ở tầng governance.
• Mẫu hiệu quả: validate bằng Pydantic rồi mới xử lý bằng Polars/DuckDB — an toàn ở cổng, tốc độ ở khâu tính; tách riêng dòng lỗi thay vì đổ vỡ cả lô.
• Cạm bẫy: coercion khoan dung có thể che dữ liệu lệch (dùng strict mode khi cần chặt); đừng dựng model cho từng dòng ở quy mô hàng chục triệu; kiểm tĩnh và validate runtime là hai lớp bổ sung, không thay thế nhau.

Nguồn tham khảo

• Pydantic Documentation — tài liệu chính thức Pydantic v2 (BaseModel, Field, validators, model_validate/model_dump, JSON Schema)
• pydantic-settings Documentation — cấu hình từ biến môi trường / file .env
• Python Documentation — typing (Support for type hints) — Literal, TypedDict, Protocol, TypeVar, Optional/Union
• Python Documentation — dataclasses — @dataclass trong thư viện chuẩn
• mypy Documentation — kiểm kiểu tĩnh
• PEP 484 — Type Hints, PEP 544 — Protocols, PEP 585 — Type Hinting Generics In Standard Collections, PEP 604 — Allow writing union types as X | Y