2.1. Pydantic

Pydantic là một thư viện mạnh mẽ cho việc xác thực dữ liệu và quản lý thiết lập cấu trúc dữ liệu và đối tượng.

Python type hint

Về cơ bản, các ngôn ngữ lập trình được chia làm 2 loại là Statically typed và Dynamically typed

• Statically typed: khi lập trình, ta phải khai báo kiểu của biến một cách tường minh, kiểu dữ liệu của chúng phải được biết tại thời điểm biên dịch (compile time)

• Dynamically typed: kiểu của biến được liên kết với giá trị tại thời điểm chạy (run time), không cần khai báo kiểu của một biến khi lập trình, nó sẽ tự động ép kiểu khi nhận được giá trị.

Python thuộc kiểu Dynamically typed. Mục đích của type hint là cung cấp một cú pháp tiêu chuẩn cho chú thích kiểu của dữ liệu đầu vào và đầu ra. Type hint giúp thông báo tới người đọc và IDEs về các kiểu dữ liệu mong đợi. Điều này thì được khuyến khích chứ không bắt buộc.

Và vì nó là type hint, nên nếu trong quá trình chạy pipeline mà kiểu dữ liệu không đúng theo định đạng nhưng ko gây ra error thì code vẫn được chấp nhận

def add(a: int, b: int) -> int:
    return a + b


add(1, 2)  # đúng


add(0.2, 1)  # sai kiểu dữ liệu nhưng run code không lỗi --> đúng


add("a", "b")  # sai kiểu dữ liệu nhưng run code không lỗi --> đúng

'ab'

2.1.1. Các mode/loại validators

Phân loại các loại validators trong Pydantic

có các loại validators sau:

• (A): Pydantic internal validation | validate data-type

• (B): BeforeValidator | mode = 'before'

• (C): AfterValidator | mode = 'after'

• (D): PlainValidator | mode = 'plain'

Thứ tự chạy các mode này được mô tả như sau:

• các BeforeValidators sẽ chạy theo thứ tự ngược so với khai báo trong cùng loại before

• các AfterValidators hoặc Field sẽ chạy theo thứ tự cùng với thứ tự khai báo

2.1.1.1. Pydantic internal validation | validate data-type

Là định nghĩa kiểu dữ liệu cần validate, đây là xác định kiểu dữ liệu cho bước Pydantic internal validation, khi đó dữ liệu cận phải map đúng với kiểu hoặc sau khi ép kiểu (ví dụ từ str -> int nếu được) thì dữ liệu sau khi ép kiểu sẽ phải đúng với datatype đã khai báo.

def add(
    a: int,
    b: int,
    c: bool,
    d: float,
    e: list,
    f: dict,
    g: set,
    h: list[str],  # each item in list is string
    i: tuple[
        int, str, float
    ],  # a tuple with 3 items, an int, another int, and a str
    k: set[bytes],  #  a set, and each of its items is of type bytes
    l: dict[str, float],  # a dict with key is string and value is float
    m: str | int,  # a value type is string or integer
    n: str | None,  # string or None
) -> int:
    return int(a) + b

Sử dụng type-hint để định nghĩa data-type của dữ liệu sau khi chạy qua các validators mode = before, validator này thực hiện việc parsing và xử lý dữ liệu và xác thực datatype. Nếu dữ liệu khớp với data-type khai báo hoặc sau khi ép kiểu (chuyển int –> str) khớp với data-type khai báo thì sẽ success hoặc không thì báo ValidationError

• Loại validator này chạy sau khi chạy hết các mode = before hoặc không có bất cứ 1 mode = plain nào trong cả pipeline validation.

• Sau khi chạy xong validator này thì sẽ chạy đến các mode = after

from pydantic import BaseModel


class MyModel(BaseModel):
    number: str  # (Pydantic internal validation)


try:
    MyModel(number=78.2981)
except Exception as e:
    print(e)

1 validation error for MyModel
number
  Input should be a valid string [type=string_type, input_value=78.2981, input_type=float]
    For further information visit https://errors.pydantic.dev/2.7/v/string_type

2.1.1.2. BeforeValidator | mode = 'before'

Là việc ta sẽ tự định nghĩa các hàm phục vụ các mục đích xử lý dữ liêu tho đầu vào hoặc làm sạch và kiểm tra trước khi thực hiện validate data-type (Pydantic internal validation).

• Trong quá trình validate giữa nhiều bước cùng mode before, validator nào đươc khai báo sau sẽ chạy trước, còn khai báo trước sẽ chạy sau (bị ngược thứ tự)

• Trong quá trình validate giữa nhiều validators khác mode: mode before sẽ chạy đầu tiên cho tới khi gặp Pydantic internal validation hoặc mode = plain

from typing import Any
from pydantic import BaseModel, BeforeValidator, Field
from typing import Annotated


def str_to_float(v: Any) -> float:
    print("str_to_float: Bỏ ký tự đầu tiên và chuyển thành float")
    v_rm0ind = v[1:]
    return float(v_rm0ind)


def float_to_int_to_str(v: float) -> str:
    print("float_to_int_to_str: chuyển về int sau đó chuyển về string")
    assert v > 0, "Value must be positive"
    return str(int(v))


class MyModel(BaseModel):
    number: Annotated[
        int,  # run #3 : (Pydantic internal validation) run sau khi chạy hết các mode = 'before'
        BeforeValidator(
            float_to_int_to_str
        ),  # run #2 : (mode 'before') khai báo mode = 'before' đầu tiên nên chạy cuối cùng trong các mode = 'before'
        BeforeValidator(
            str_to_float
        ),  # run #1 : (mode 'before') do khai báo mode = 'before' sau cùng nên chạy đầu tiên của pipeline
    ]


MyModel(number="78.2981")

str_to_float: Bỏ ký tự đầu tiên và chuyển thành float
float_to_int_to_str: chuyển về int sau đó chuyển về string

MyModel(number=8)

2.1.1.3. AfterValidator | mode = 'after'

Là việc ta sẽ tự định nghĩa các hàm phục vụ các mục đích xử lý dữ liêu làm việc với dữ liệu đã được xác thực và chuyển đổi thành đúng kiểu dữ liệu mong muốn sau khi thực hiện validate bằng Pydantic internal validation.

• Trong quá trình validate giữa nhiều bước cùng mode after, validator nào đươc khai báo trước sẽ chạy trước, còn khai báo sau sẽ chạy sau (theo đúng thứ tự khai báo)

• Trong quá trình validate giữa nhiều validators khác mode: mode after sẽ chạy sau khi chạy Pydantic internal validation hoặc mode = plain

Field | validate data-value : là 1 dạng after validators được khởi tạo nhanh để kiểm tra những method đơn giản của dữ liệu

from typing import Any
from pydantic import BaseModel, BeforeValidator, AfterValidator
from typing import Annotated


def str_to_float(v: Any) -> float:
    print("str_to_float: Bỏ ký tự đầu tiên và chuyển thành float")
    v_rm0ind = v[1:]
    return float(v_rm0ind)


def float_to_int_to_str(v: float) -> str:
    print("float_to_int_to_str: chuyển về int sau đó chuyển về string")
    assert v > 0, "Value must be positive"
    return str(int(v))


def double(v: Any) -> Any:
    print("double: int sau đó nhân đôi")
    return (10 + v) * 2


def check_squares(v: int) -> int:
    print("check_squares: kiểm tra xem có phải số chính phương")
    assert v**0.5 % 1 == 0, f"{v} is not a square number"
    return v


class MyModel(BaseModel):
    number: Annotated[
        int,  # run #3 : (Pydantic internal validation) run sau khi chạy hết các mode = 'before'
        BeforeValidator(
            float_to_int_to_str
        ),  # run #2 : (mode 'before') khai báo mode = 'before' đầu tiên nên chạy cuối cùng trong các mode = 'before'
        BeforeValidator(
            str_to_float
        ),  # run #1 : (mode 'before') do khai báo mode = 'before' sau cùng nên chạy đầu tiên của pipeline
        AfterValidator(
            double
        ),  # run #4 : (mode 'after') khai báo mode = 'after' đầu tiên nên chạy đầu tiên trong các mode = 'after' nhưng sau data-type
        Field(
            gt=0, lt=100
        ),  # run #5 : (mode 'after') là 1 dạng after validator
        AfterValidator(
            check_squares
        ),  # run #6 : (mode 'after') khai báo mode = 'after' cuôi cùng nên chạy cuối cùng của pipeline
    ]


MyModel(number="78.2981")

str_to_float: Bỏ ký tự đầu tiên và chuyển thành float
float_to_int_to_str: chuyển về int sau đó chuyển về string
double: int sau đó nhân đôi
check_squares: kiểm tra xem có phải số chính phương

MyModel(number=36)

2.1.1.3.1. Field | validate data-value

Sử dụng Field để định nghĩa giá trị data-value hợp lệ của dữ liệu là 1 dạng validator mode = after, thường được sử dụng kết hợp với Annotated

Field parameters:

• default : default value

• alias : đặt biệt danh cho trường thông tin khác với tên trong trường dữ liệu được xác thực

• validation_alias : đặt biệt danh cho trường thông tin giống alias nhưng chỉ sử dụng only for validation

• serialization_alias được sử dụng khi tên trường nội bộ của mô hình giống alias, nhưng không nhất thiết phải là tên bạn muốn sử dụng khi tuần tự hóa mô hình

• strict: cho phép được sử dụng trong strict mode = không được phép ép kiểu ? (xem bảng bảng tham chiếu ep kiểu table)

• exclude : can be used to control which fields should be excluded from the model when exporting the model

• description: mô tả Field

• validate_default: bool = False: có validate giá trị mặc định của field hay không ? (vì ta thường giả sử là giá trị mặc định sẽ luôn thoả mãn validation)

tham số cho Numeric

• gt : greater than

• lt : less than

• ge : greater than or equal to

• le : less than or equal to

• multiple_of : a multiple of the given number

• allow_inf_nan : allow ‘inf’, ‘-inf’, ‘nan’ values

tham số cho Decimal

• max_digits : Maximum number of digits within the Decimal. It does not include a zero before the decimal point or trailing decimal zeroes.

• decimal_places : Maximum number of decimal places allowed. It does not include trailing decimal zeroes.

tham số cho String

• min_length : Minimum length of the string.

• max_length : Maximum length of the string.

• pattern : A regular expression that the string must match.

tham số cho Dataclass

• init: Whether the field should be included in the init of the dataclass.

• init_var: Whether the field should be seen as an init-only field in the dataclass.

• kw_only: Whether the field should be a keyword-only argument in the constructor of the dataclass.

tham số cho Discriminator: phân loại

• discriminator : sử dụng khi muốn control giá trị chỉ nằm trong 1 list các giá trị từ 1 attribute từ 1 số class nào đó

from typing import Literal
from pydantic import BaseModel, Field

class Cat(BaseModel):
    pet_type: Literal["cat"]
    age: int

class Dog(BaseModel):
    pet_type: Literal["dog"]
    age: int

class Model(BaseModel):
    # pet chỉ nhật được trong các giá trị là 'cat' hoặc 'dog' là attribute pet_type của các class định nghĩa
    pet: Cat | Dog = Field(discriminator="pet_type")

from pydantic import HttpUrl, PastDate
from pydantic import Field
from pydantic import validate_call
from typing import Annotated
from uuid import uuid4


@validate_call(validate_return=True)
def process_payload(
    url: HttpUrl,  # là dạng url hyperlink
    name: Annotated[
        str, Field(min_length=2, max_length=15)
    ],  # string có độ dài ký tự là 2 --> 15
    birth_date: PastDate,  # là date nhưng là giá trị quá khứ
    id: str = Field(
        default_factory=lambda: uuid4().hex
    ),  # sử dụng trực tiếp Field
) -> str:
    return url, name, birth_date, id


payload = {
    "url": "httpss://example.com",
    "name": "A",
    "birth_date": "2024-12-12",
    "id": "100017727",
}
try:
    process_payload(**payload)
except Exception as e:
    print(e)

3 validation errors for process_payload
url
  URL scheme should be 'http' or 'https' [type=url_scheme, input_value='httpss://example.com', input_type=str]
    For further information visit https://errors.pydantic.dev/2.9/v/url_scheme
name
  String should have at least 2 characters [type=string_too_short, input_value='A', input_type=str]
    For further information visit https://errors.pydantic.dev/2.9/v/string_too_short
birth_date
  Date should be in the past [type=date_past, input_value='2024-12-12', input_type=str]
    For further information visit https://errors.pydantic.dev/2.9/v/date_past

2.1.1.4. PlainValidator | mode = 'plain'

Là việc ta sẽ tự định nghĩa các hàm phục vụ các mục đích bao gồm chức năng của cả validator mode before và Pydantic internal validation (bao gồm xử lý dữ liêu thô và chuyển đổi thành đúng kiểu dữ liệu mong muốn, sau đó xác thực data-type). Cho nên khi chạy qua 1 validator mode = plain thì sẽ không chạy qua thêm bất cứ 1 validator mode before và Pydantic internal validation nào nữa.

• Trong quá trình validate nếu có mode plain thì chỉ có 1 validator mode plain, về thứ tự tương đương với 1 validator mode before, tức là có thể chạy qua các mode before phía trước (khai báo sau), sau khi chạy qua validator mode plain sẽ đến luồng mode after

from typing import Any
from pydantic import BaseModel, BeforeValidator, AfterValidator, PlainValidator
from typing import Annotated


def str_to_float(v: Any) -> float:
    print("str_to_float: Bỏ ký tự đầu tiên và chuyển thành float")
    v_rm0ind = v[1:]
    return float(v_rm0ind)


def float_to_int_to_str(v: float) -> str:
    print("float_to_int_to_str: chuyển về int sau đó chuyển về string")
    assert v > 0, "Value must be positive"
    return str(int(v))


def check_is_float(v: Any) -> Any:
    print("check_is_float: Plain validator")
    assert isinstance(v, float), f"{v} is not float"
    return v


def double(v: Any) -> Any:
    print("double: int sau đó nhân đôi")
    return (10 + v) * 2


def check_squares(v: int) -> int:
    print("check_squares: kiểm tra xem có phải số chính phương")
    assert v**0.5 % 1 == 0, f"{v} is not a square number"
    return v


class MyModel(BaseModel):
    number: Annotated[
        str,  # not run : (Pydantic internal validation) ko run do trong pipeline chứa mode 'plain'
        BeforeValidator(
            float_to_int_to_str
        ),  # not run : (mode 'before') khai báo mode = 'before' trước mode = 'plain' nên sẽ không run
        PlainValidator(
            check_is_float
        ),  # run #2 : (mode 'plain') khai báo mode = 'plain' nên thứ tự chạy tương đương với validator mode 'before'
        BeforeValidator(
            str_to_float
        ),  # run #1 : (mode 'before') do khai báo mode = 'before' sau cùng và sau mode 'plain' nên chạy đầu tiên của pipeline
        AfterValidator(
            double
        ),  # run #3 : (mode 'after') khai báo mode = 'after' đầu tiên nên chạy đầu tiên trong các mode = 'after' nhưng sau data-type
        Field(
            gt=0, lt=100
        ),  # run #4 : (mode 'after') là 1 dạng after validator
        AfterValidator(
            check_squares
        ),  # run #5 : (mode 'after') khai báo mode = 'after' cuôi cùng nên chạy cuối cùng của pipeline
    ]


MyModel(number="78.0")

str_to_float: Bỏ ký tự đầu tiên và chuyển thành float
check_is_float: Plain validator
double: int sau đó nhân đôi
check_squares: kiểm tra xem có phải số chính phương

MyModel(number=36.0)

2.1.2. Cách định nghĩa validators

Định nghĩa validators giúp xác định xem 1 field sẽ run qua pipeline validation theo phương pháp nào

2.1.2.1. Định nghĩa bằng Annotated

Cấu trúc:

Annotated[<data-type>, <Các loại validators khác>]

Thứ tự khai báo cái validators theo thứ tự khai báo trong list Annotated (chú ý là với validator mode before thì thứ tự run ngược lại với thứ tự khai báo)

Sử dụng cho class validation

from typing import Any
from pydantic import BaseModel, BeforeValidator, AfterValidator
from typing import Annotated


def str_to_float(v: Any) -> float:
    print("str_to_float: Bỏ ký tự đầu tiên và chuyển thành float")
    v_rm0ind = v[1:]
    return float(v_rm0ind)


def float_to_int_to_str(v: float) -> str:
    print("float_to_int_to_str: chuyển về int sau đó chuyển về string")
    assert v > 0, "Value must be positive"
    return str(int(v))


def double(v: Any) -> Any:
    print("double: int sau đó nhân đôi")
    return (10 + v) * 2


def check_squares(v: int) -> int:
    print("check_squares: kiểm tra xem có phải số chính phương")
    assert v**0.5 % 1 == 0, f"{v} is not a square number"
    return v


MyNumber = Annotated[
    int,  # run #3 : (Pydantic internal validation) run sau khi chạy hết các mode = 'before'
    BeforeValidator(
        float_to_int_to_str
    ),  # run #2 : (mode 'before') khai báo mode = 'before' đầu tiên nên chạy cuối cùng trong các mode = 'before'
    BeforeValidator(
        str_to_float
    ),  # run #1 : (mode 'before') do khai báo mode = 'before' sau cùng nên chạy đầu tiên của pipeline
    AfterValidator(
        double
    ),  # run #4 : (mode 'after') khai báo mode = 'after' đầu tiên nên chạy đầu tiên trong các mode = 'after' nhưng sau data-type
    Field(gt=0, lt=100),  # run #5 : (mode 'after') là 1 dạng after validator
    AfterValidator(
        check_squares
    ),  # run #6 : (mode 'after') khai báo mode = 'after' cuôi cùng nên chạy cuối cùng của pipeline
]


class MyModel(BaseModel):
    numbers: list[MyNumber]


MyModel(numbers=["78.2981"])

str_to_float: Bỏ ký tự đầu tiên và chuyển thành float
float_to_int_to_str: chuyển về int sau đó chuyển về string
double: int sau đó nhân đôi
check_squares: kiểm tra xem có phải số chính phương

MyModel(numbers=[36])

Sử dụng trong function validation

from pydantic import validate_call, Field
from typing import Annotated


@validate_call
def how_many(num: Annotated[int, Field(gt=10)]):
    return num


try:
    how_many(9)
except Exception as e:
    print(e)

1 validation error for how_many
0
  Input should be greater than 10 [type=greater_than, input_value=9, input_type=int]
    For further information visit https://errors.pydantic.dev/2.7/v/greater_than

Sử dụng trực tiếp Annotated

from typing import Annotated
from pydantic import Field

my = Annotated[
    int,  # RUN #3
    Field(gt=0, lt=100),  # RUN #4
    Field(multiple_of=4),  # RUN #5
]

my(78.34)

78

2.1.2.2. Decorate bằng field_validator

@field_validator là một decorator được sử dụng để định nghĩa các hàm (validators) tùy chỉnh nhằm xác thực (validate) và xử lý các trường dữ liệu (fields) của mô hình (model). @field_validator cho phép bạn xác định các quy tắc hoặc logic xác thực đặc biệt để kiểm tra giá trị của các trường trước khi chúng được gán vào model.

Chú ý: @field_validator nên decorate cho 1 classmethod @classmethod

from pydantic import BaseModel, field_validator

class MyModel(BaseModel):
    my_field: str

    @field_validator('my_field')
    def validate_my_field(cls, v, info: ValidationInfo ):
        # logic xác thực
        return v

Tham số của field_validator

• mode = before | after | plain

• check_fields : check xem field đó có thực sự tồn tại hay không ?

Tham số của hàm validator

• v: is the field value to validate

• info: is an instance of pydantic.ValidationInfo

Thứ tự khai báo cái validators cho mỗi 1 trường sẽ theo thứ tự xuất hiện cho validator trong class (chú ý là với validator mode before thì thứ tự run ngược lại với thứ tự khai báo)

from pydantic import BaseModel, field_validator, ValidationInfo
from typing import Annotated


class MyModel(BaseModel):
    numbers: Annotated[
        int,  # RUN #3
        Field(gt=0, lt=100),  # RUN #4
        Field(multiple_of=4),  # RUN #5
    ]

    # RUN #2: chạy thứ 2
    @field_validator("numbers", mode="before")  # validate field : numbers
    @classmethod
    def float_to_int_to_str(cls, v: float, info: ValidationInfo) -> str:
        print("float_to_int_to_str: chuyển về int sau đó chuyển về string")
        assert v > 0, "Value must be positive"
        return str(int(v))

    # RUN #1: chạy đầu tiên
    @field_validator("numbers", mode="before")  # validate field : numbers
    @classmethod
    def str_to_float(cls, v: Any, info: ValidationInfo) -> float:
        print("str_to_float: Bỏ ký tự đầu tiên và chuyển thành float")
        v_rm0ind = v[1:]
        return float(v_rm0ind)

    # RUN #6
    @field_validator("*", mode="after")  # validate all field
    @classmethod
    def double(cls, v: Any, info: ValidationInfo) -> Any:
        print("double: int sau đó nhân đôi")
        return (10 + v) * 2

    # RUN #7
    @field_validator("*", mode="after")  # validate all field
    @classmethod
    def check_squares(cls, v: int, info: ValidationInfo) -> int:
        print("check_squares: kiểm tra xem có phải số chính phương")
        assert v**0.5 % 1 == 0, f"{v} is not a square number"
        return v


MyModel(numbers="78.2981")

str_to_float: Bỏ ký tự đầu tiên và chuyển thành float
float_to_int_to_str: chuyển về int sau đó chuyển về string
double: int sau đó nhân đôi
check_squares: kiểm tra xem có phải số chính phương

MyModel(numbers=36)

2.1.2.3. Decorate bằng model_validator

Sử dụng model_validator nếu muốn validate bằng mối quan hệ giữa nhiều trường field với nhau (thay vì chỉ validate 1 trường duy nhất)

thứ tự chạy : @model_validator(mode = 'before') —> @field_validator —> @model_validator(mode = 'after')

model_validator sẽ bọc ngoài field_validator

from typing import Any
from typing_extensions import Self
from pydantic import BaseModel, model_validator, field_validator


class UserModel(BaseModel):
    username: str
    password1: str
    password2: str

    @field_validator("password1", "password2", mode="after")
    @classmethod
    def strip_data(cls, v):
        return str(v).strip()

    # kiểm tra input đầu vào không nên có trường card_number
    # sử dụng với classmethod và lấy data từ argument đầu tiên : data
    @model_validator(mode="before")
    @classmethod
    def check_card_number_omitted(cls, data: Any) -> Any:
        if isinstance(data, dict):
            assert (
                "card_number" not in data
            ), "card_number should not be included"
        return data

    # kiểm tra password1 trùng với password2
    # sử dụng trong self và lấy data từ self
    @model_validator(mode="after")
    def check_passwords_match(self) -> Self:
        pw1 = self.password1
        pw2 = self.password2
        if pw1 is not None and pw2 is not None and pw1 != pw2:
            raise ValueError("passwords do not match")
        return self


# lỗi khi ở before validator
try:
    u = UserModel(
        username="scolvin",
        password1="zxcvbn",
        password2="  zxcvbn ",
        card_number="1234",
    )
    print(u)
except Exception as e:
    print(e)

1 validation error for UserModel
  Assertion failed, card_number should not be included [type=assertion_error, input_value={'username': 'scolvin', '..., 'card_number': '1234'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.9/v/assertion_error

# lỗi khi ở after validator
try:
    UserModel(
        username="scolvin",
        password1="zxcvbn",
        password2="zxcvbn-123",
    )
except Exception as e:
    print(e)

1 validation error for UserModel
  Value error, passwords do not match [type=value_error, input_value={'username': 'scolvin', '...assword2': 'zxcvbn-123'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.9/v/value_error

2.1.3. Các cách áp dụng Validation

Các áp dụng Pydantic validation trong python

2.1.3.1. Validation for function (def) - validate_call

validate_call để xác thực các lần gọi hàm dựa trên type hint và chú thích (annotations)

• Pydantic được sử dụng để đảm bảo dữ liệu bạn truyền vào mỗi lần gọi hàm tuân theo các kiểu dữ liệu mong đợi (type hints).

• Pydantic sẽ cố gắng hiểu ý của bạn và sử dụng ép kiểu (type coercion). Điều này có thể dẫn đến việc các giá trị string được truyền vào lần gọi hàm bị chuyển đổi ngầm thành kiểu dữ liệu mong đợi.

  Ép kiểu không phải lúc nào cũng khả thi, và trong trường hợp đó, ValidationError sẽ được phát sinh.

import pydantic


@pydantic.validate_call
def add(a: int, b: int) -> int:
    return a + b


add(2, 3)

5

báo lỗi khi sai kiểu dữ liệu input

add("a", 3)

ValidationError: 1 validation error for add 0

Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value=’a’, input_type=str] For further information visit https://errors.pydantic.dev/2.7/v/int_parsing

sẽ đúng nếu ép kiểu thành công

add("3", "3")

6

Lỗi nếu ép kiểu từ float sang int mà giá trị bị thay đổi:

• 3.0 –> int (OKE)

• 0.3 –> int (bị thay đổi)

add(3.0, 0.3)

Input should be a valid integer, got a number with a fractional part [type=int_from_float, input_value=0.3, input_type=float] For further information visit https://errors.pydantic.dev/2.7/v/int_from_float

validate_return=True nếu muốn validate giá trị trả về, (mặc định là False), phướng pháp ép kiểu cũng được áp dụng

ví dụ output của hàm định nghĩa là int, tuy nhiên output thực tế là string dạng int thì vẫn OKE

from pydantic import validate_call


@validate_call(validate_return=True)
def add(*args: int, a: int, b: int = 4) -> int:
    return str(sum(args) + a + b)


add(4, 3, 4, a=10)

25

Customize the validator

from pydantic import validate_call
from datetime import datetime


@validate_call
def how_many(num: Annotated[int, Field(gt=10)]):
    return num


@validate_call
def when(dt: datetime = Field(default_factory=datetime.now)):
    return dt

2.1.3.2. Validation for class - dataclass

Sử dụng decorator pydantic.dataclasses.dataclass nếu chỉ cần validate đơn giản và ưu tiên performance, áp dụng cho class

2.1.3.3. Validation for class - BaseModel

Kế thừa từ BaseModel, chúng ta sử dụng một cơ chế ẩn để thực hiện việc xác thực dữ liệu nếu cần validate cả data-type và data-value, phân tích cú pháp, và tuần tự hóa.

–> Điều này cho phép chúng ta tạo ra các đối tượng phù hợp với các thông số của mô hình.

from pydantic import Field
from pydantic import BaseModel
from typing import Annotated


class Person(BaseModel):
    name: str = Field(min_length=2, max_length=15)

    # cách khác sử dụng Annotated
    age: Annotated[int, Field(default=1, ge=0, le=120)]


# ----

john = Person(name="john", age=20)
print(john)
print("---------------------")

# ----
try:
    mike = Person(name="m", age=0)
except Exception as e:
    print(e)

name='john' age=20
---------------------
2 validation errors for Person
name
  String should have at least 2 characters [type=string_too_short, input_value='m', input_type=str]
    For further information visit https://errors.pydantic.dev/2.7/v/string_too_short
age
  Input should be greater than 0 [type=greater_than, input_value=0, input_type=int]
    For further information visit https://errors.pydantic.dev/2.7/v/greater_than

2.1.3.3.1. validation for property attribute

Sử dụng property trong class thường được sử dụng khi tính toán 1 trường thông tin từ các trường khác hoặc việc tính toán trở nên phức tạp hoặc tốn nhiều chi phí, do đó các attribute này thường được tính khi call hoặc khi cần.

• computed_field áp dụng cho các atrribute trong class được tính toán theo property nhưng vẫn đảm bảo sẽ không tính toán các attribute này cho đến khi được call

from pydantic import BaseModel, computed_field


class Box(BaseModel):
    width: float
    height: float
    depth: float

    @computed_field
    @property  # attribute này sẽ không được tính toán cho đến khi call
    def volume(self) -> float:
        print("đã chạy dòng code tính attribute volume")
        return self.width * self.height * self.depth


b = Box(width=1, height=2, depth=3)

volume vẫn được validate datatype khi được call

print(b.model_dump())

đã chạy dòng code tính attribute volume
{'width': 1.0, 'height': 2.0, 'depth': 3.0, 'volume': 6.0}

2.1.3.3.2. model_validate(): xác thực dữ liệu từ 1 object

tương tụ như __init__ method của model (khi tạo đối tượng bằng việc khai báo arguments class đó), thay vào đó là truyền vào dict hoặc object

• Chức năng: Xác thực (validate) một đối tượng dựa trên mô hình Pydantic đã được định nghĩa. Nó kiểm tra dữ liệu có đúng kiểu và giá trị hay không dựa trên các quy tắc xác thực (validation rules) mà bạn đã định nghĩa trong mô hình.

• Ứng dụng: Sử dụng khi bạn muốn đảm bảo rằng dữ liệu đầu vào của mình khớp với các yêu cầu của mô hình trước khi xử lý thêm.

from datetime import datetime
from typing import Optional

from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str = "John Doe"
    signup_ts: Optional[datetime] = None


u = User.model_validate(
    {"id": 123, "name": "James"}
)  # --> return object of class
u

User(id=123, name='James', signup_ts=None)

• model_validate_json(): this validates the provided data as a JSON string or bytes object. (nếu data đang dưới dạng JSON hoặc bytes thì cách này sẽ nhanh hơn)

2.1.3.3.3. model_construct(): nhận dữ liệu bỏ qua validation

This can be useful in at least a few cases:

• data phức tạp được truyền vào nhưng đã biết được datatype, nên ko cần validate để đảm bảo performance

• khi đã có 1 số hàm validate phía trong và không cần trigger validate lại lần nữa

User.model_construct({"id": "ạkshdajs", "name": [1231231]})

User(name='John Doe', signup_ts=None)

2.1.3.3.4. model_dump(): convert to dict

• Chức năng: Trả về một từ điển chứa các trường (fields) và giá trị (values) tương ứng của mô hình.

• Ứng dụng: Khi bạn muốn tuần tự hóa (serialize) mô hình thành một dạng dễ xử lý hơn, ví dụ như trước khi lưu vào cơ sở dữ liệu hoặc trả về qua API.

u.model_dump()

{'id': 123, 'name': 'James', 'signup_ts': None}

• model_dump_json(): Trả về chuỗi JSON đại diện cho dữ liệu của mô hình (thực chất là kết quả của model_dump() nhưng dưới dạng JSON)

u.model_dump_json()

'{"id":123,"name":"James","signup_ts":null}'

2.1.3.3.5. model_copy() : copy model

• Chức năng: Trả về một bản sao của mô hình.

  • Mặc định là shallow copy (bản sao nông, nghĩa là chỉ sao chép đối tượng gốc mà không sao chép sâu các thành phần con bên trong).

  • Có thể chỉ định để tạo deep copy (bản sao sâu).

• Ứng dụng: Khi bạn cần một bản sao của mô hình để thực hiện các thay đổi mà không làm ảnh hưởng đến bản gốc.

2.1.3.3.6. model_json_schema() : trả về dictionary of class

• Chức năng: Trả về một từ điển có thể chuyển đổi thành JSON, đại diện cho lược đồ JSON (JSON Schema) của mô hình. JSON Schema cung cấp thông tin chi tiết về kiểu dữ liệu, các quy tắc xác thực, v.v.

• Ứng dụng: Khi bạn cần cung cấp tài liệu cho API hoặc các dịch vụ khác về cấu trúc dữ liệu được mong đợi.

u.model_json_schema()

{'properties': {'id': {'title': 'Id', 'type': 'integer'},
  'name': {'default': 'John Doe', 'title': 'Name', 'type': 'string'},
  'signup_ts': {'anyOf': [{'format': 'date-time', 'type': 'string'},
    {'type': 'null'}],
   'default': None,
   'title': 'Signup Ts'}},
 'required': ['id'],
 'title': 'User',
 'type': 'object'}

2.1.3.3.7. model_post_init() : chạy hàm sau khi __init__

• Chức năng: Phương thức này thực hiện các hành động bổ sung sau khi mô hình đã được khởi tạo (sau khi quá trình khởi tạo và xác thực hoàn tất).

• Ứng dụng: Sử dụng để thực hiện các thao tác tùy chỉnh sau khi mô hình đã được khởi tạo, như tính toán thêm các giá trị dựa trên các trường đã xác định.

class User(BaseModel):
    id: int
    name: str = "John Doe"
    signup_ts: Optional[datetime] = None

    def model_post_init(self, *args, **kwargs):
        # Tính toán giá trị tổng sau khi mô hình đã được khởi tạo
        print("AAAAAAAAA")

    @computed_field
    @property
    def name_id(self) -> str:
        return f"{self.name}-{self.id}"


u = User(id=123, name="James")

AAAAAAAAA

2.1.3.3.8. model_fields : trả về danh sách định nghĩa các field

• Chức năng: Một ánh xạ (mapping) giữa tên trường (field names) và các định nghĩa của chúng (FieldInfo instances) (not Value). Mỗi trường trong mô hình đều có một định nghĩa riêng về kiểu dữ liệu và các quy tắc xác thực.

• Ứng dụng: Sử dụng khi bạn cần truy cập hoặc kiểm tra thông tin về các trường trong mô hình (như kiểu dữ liệu, giá trị mặc định, v.v.).

Chú ý:

• trả về là định nghĩa field, chứ ko phải value

• không bao gồm các computed_fields / property

u.model_fields

{'id': FieldInfo(annotation=int, required=True),
 'name': FieldInfo(annotation=str, required=False, default='John Doe'),
 'signup_ts': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None)}

2.1.3.3.9. model_computed_fields : trả về danh sách định nghĩa các field được tính toán

• Chức năng: Ánh xạ giữa tên các trường tính toán (computed field names) và các định nghĩa của chúng (ComputedFieldInfo instances). Các trường này không được cung cấp trực tiếp mà được tính toán dựa trên các trường khác.

• Ứng dụng: Khi bạn muốn kiểm tra hoặc làm việc với các trường tính toán trong mô hình.

u.model_computed_fields

{'name_id': ComputedFieldInfo(wrapped_property=<property object at 0x000001AFEC920CC0>, return_type=<class 'str'>, alias=None, alias_priority=None, title=None, description=None, deprecated=None, examples=None, json_schema_extra=None, repr=True)}

2.1.3.3.10. model_fields_set : kiểm trả giá trị của trường được khia báo hay mặc định

• Chức năng: Tập hợp các trường (fields) đã được cung cấp rõ ràng khi mô hình được khởi tạo, có nghĩa là danh sách các trường đã được truyền input vào thay vì lấy giá trị mặc định

• Ứng dụng: Dùng để kiểm tra xem trường nào đã được gán giá trị ban đầu và trường nào có giá trị mặc định.

u.model_fields_set

{'id', 'name'}

2.1.3.3.11. model_config: thay đổi tính chất của class

Sử dụng ConfigDict để sử đổi attribute model_config của class, từ đó sửa đổi các tính chất của class:

• use_enum_values=True : khi call model lấy giá trị thì sẽ trả về value của model (.value) thay vì để dạng raw là <LLMProviders.OPENAI: 'openai'>

from pydantic import Field
from pydantic import BaseModel
from typing import Literal
from enum import Enum


# định nghĩa LLMProviders là dạng string chỉ được nhận 1 trong các giá trị sau : openai, claude.
# Thay vì sử dụng Literal định nghĩa trong attribute
class LLMProviders(str, Enum):
    OPENAI = "openai"
    CLAUDE = "claude"


class LLMParams(BaseModel):
    temperature: int = Field(validation_alias="llm_temperature", ge=0, le=1)
    llm_name: str = Field(
        validation_alias="llm_model_name", serialization_alias="model"
    )


class Payload(BaseModel):
    req_id: str = Field(exclude=True, min_length=2, max_length=15)
    text: str = Field(min_length=5)
    instruction: Literal["embed", "chat"]
    llm_provider: LLMProviders
    llm_params: LLMParams


payload = {
    "req_id": "test",
    "text": "This is a sample text.",
    "instruction": "embed",
    "llm_provider": "openai",
    "misc": "what",
    "llm_params": {"llm_temperature": 0, "llm_model_name": "gpt4o"},
}
validated_payload = Payload(**payload)
print(validated_payload)

req_id='test' text='This is a sample text.' instruction='embed' llm_provider=<LLMProviders.OPENAI: 'openai'> llm_params=LLMParams(temperature=0, llm_name='gpt4o')

validated_payload.model_dump()

{'text': 'This is a sample text.',
 'instruction': 'embed',
 'llm_provider': <LLMProviders.OPENAI: 'openai'>,
 'llm_params': {'temperature': 0, 'llm_name': 'gpt4o'}}

validated_payload.model_dump(by_alias=True)

{'text': 'This is a sample text.',
 'instruction': 'embed',
 'llm_provider': <LLMProviders.OPENAI: 'openai'>,
 'llm_params': {'temperature': 0, 'model': 'gpt4o'}}

Tiến hành sửa lại model_config thông qua class ConfigDict

from pydantic import ConfigDict


class Payload(BaseModel):
    req_id: str = Field(exclude=True)
    text: str = Field(min_length=5)
    instruction: Literal["embed", "chat"]
    llm_provider: LLMProviders
    llm_params: LLMParams

    # thêm trường model_config
    model_config = ConfigDict(
        use_enum_values=True
    )  # khi call sẽ trả ra class.value thay vì class: <LLMProviders.OPENAI: 'openai'> sẽ chuyển thành 'openai'


validated_payload = Payload(**payload)
validated_payload.model_dump(by_alias=True)

{'text': 'This is a sample text.',
 'instruction': 'embed',
 'llm_provider': 'openai',
 'llm_params': {'temperature': 0, 'model': 'gpt4o'}}

2.1.4. Specific Types in Pydantic

Pydantic cung cấp rất nhiều các format định dạng các object khác nhau để validate

• Định dạng datatype trong python

• Định dạng đối tượng ngoài thực tế: people, network, phone, …

• …

Check : https://docs.pydantic.dev/latest/api/types/

2.1.5. Use-case

2.1.5.1. Validate dataframe from csv

Mô tả dữ liệu data in csv file:

name,age,bank_account
johnny,0,20
matt,10,0
abraham,100,100000
mary,15,15
linda,130,100000

Sử dụng thư viện pandantic được fork lại pydantic bổ sung thêm method parse_df cho BaseModel, hoặc ta có thể tự build lại theo code phía dưới

"""A subclass of the Pydantic BaseModel that adds a parse_df method to validate DataFrames."""
from __future__ import annotations

import logging
import math
import os
from typing import Any, Literal

import pandas as pd
from multiprocess import (  # type:ignore # pylint: disable=no-name-in-module
    Process,
    Queue,
)
from pydantic import BaseModel


class PandanticBaseModel(BaseModel):
    """A subclass of the Pydantic BaseModel that adds a parse_df method to validate DataFrames."""

    @classmethod
    def parse_df(
        cls,
        dataframe: pd.DataFrame,
        errors: Literal['raise', 'filter'] = "raise",
        context: dict[str, Any] | None = None,
        n_jobs: int = 1,
        verbose: bool = True,
    ) -> pd.DataFrame:
        """Validate a DataFrame using the schema defined in the Pydantic model.

        Args:
            dataframe (pd.DataFrame): The DataFrame to validate.
            errors (str, optional): How to handle validation errors. Defaults to "raise".
            context (Optional[dict[str, Any], None], optional): The context to use for validation.
            n_jobs (int, optional): The number of processes to use for validation. Defaults to 1.
            verbose (bool, optional): Whether to log validation errors. Defaults to True.

        Returns:
            pd.DataFrame: The DataFrame with valid rows in case of errors="filter".
        """
        errors_index = []
        logging.debug("Amount of available cores: %s", os.cpu_count())

        dataframe = dataframe.copy()
        dataframe["_index"] = dataframe.index

        if n_jobs != 1:
            if n_jobs < 0:
                n_jobs = os.cpu_count()  # type: ignore

            chunks = []
            chunk_size = math.floor(len(dataframe) / n_jobs)
            num_chunks = len(dataframe) // chunk_size + 1

            q = Queue()

            for i in range(num_chunks):
                chunks.append(dataframe.iloc[i * chunk_size : (i + 1) * chunk_size])

            for i in range(num_chunks):
                p = Process(  # pylint: disable=not-callable
                    target=cls._validate_row, args=(chunks[i], q), daemon=True
                )
                p.start()

            num_stops = 0
            for i in range(num_chunks):
                while True:
                    index = q.get()
                    if index is None:
                        num_stops += 1
                        break

                    errors_index.append(index)

                if num_stops == num_chunks:
                    break
        else:
            for row in dataframe.to_dict("records"):
                try:
                    cls.model_validate(
                        obj=row,
                        context=context,
                    )
                except Exception as exc:  # pylint: disable=broad-exception-caught
                    if verbose:
                        print(exc)
                        logging.info("Validation error found at index %s\n%s", row["_index"], exc)

                    errors_index.append(row["_index"])

        logging.debug("# invalid rows: %s", len(errors_index))

        if len(errors_index) > 0 and errors == "raise":
            raise ValueError(f"{len(errors_index)} validation errors found in dataframe.")
        if len(errors_index) > 0 and errors == "filter":
            return dataframe[~dataframe.index.isin(list(errors_index))].drop(columns=["_index"])

        return dataframe.drop(columns=["_index"])

    @classmethod
    def _validate_row(
        cls,
        chunk: pd.DataFrame,
        q: Queue,
        context: dict[str, Any] | None = None,
        verbose: bool = True,
    ) -> None:
        """Validate a single row of a DataFrame.

        Args:
            chunk (pd.DataFrame): The DataFrame chunk to validate.
            q (Queue): The queue to put the index of the row in case of an error.
            context (Optional[dict[str, Any], None], optional): The context to use for validation.
            verbose (bool, optional): Whether to log validation errors. Defaults to True.
        """
        logging.debug("Process started.")

        for row in chunk.to_dict("records"):
            try:
                cls.model_validate(
                    obj=row,
                    context=context,
                )
            except Exception as exc:  # pylint: disable=broad-exception-caught
                if verbose:
                    logging.info("Validation error found at index %s\n%s", row["_index"], exc)

                q.put(row["_index"])

        logging.debug("Process ended.")
        q.put(None)

import pandas as pd
from pandantic import BaseModel # type: ignore

FILE_NAME = 'data/testcase.csv'

class DataModel(BaseModel):
    name: str = Field(min_length=2, max_length=15)
    age: int = Field(ge=1, le=120)
    bank_account: float = Field(ge=0, default=0)

    @field_validator('name')
    @classmethod
    def validate_name(cls, v: str, info: ValidationInfo) -> str:
        return str(v).capitalize()

df = pd.read_csv(FILE_NAME)

df_validated = DataModel.parse_df(df, errors='filter', n_jobs= 1, verbose=False)

df_validated

	name	age	bank_account
1	matt	10	0
2	abraham	100	100000
3	mary	15	15

2.1.5.2. validate FastAPI request

FastAPI đã được tích hợp với Pydantic. Cách FastAPI xử lý request là chuyển chúng đến một hàm xử lý route. Bằng cách chuyển request này đến hàm, việc xác thực được thực hiện tự động, tương tự như validate_call.

Khi này, các request được tự động xác thực bằng pydantic

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

class Request(BaseModel):
    request_id: str
    url: HttpUrl

app = FastAPI()

@app.post("/search/by_url/")
async def create_item(req: Request):
    return item