Mở đầu

Giả sử chúng ta có một tệp CSV với nhiều cột và hàng nghìn dòng. Trong phân tích dữ liệu, thông thường ta sẽ tải tệp CSV này vào Pandas DataFrame để kiểm tra. Bạn có thể sẽ thực hiện kiểm tra và làm sạch dữ liệu, loại bỏ một số cột và tạo ra các cột mới. Tuy nhiên, việc này không phải lúc nào cũng rõ ràng với người khác, họ phải mở tệp CSV hoặc xem qua code để hiểu cách các cột được sử dụng và tạo ra. Điều này hữu ích cho các bước đầu của phân tích và nghiên cứu dữ liệu. Nhưng khi đã phân tích xong và cần tạo pipeline để tải, biến đổi và sử dụng dữ liệu cho mục đích cao hơn (ví dụ đưa vào mô hình học máy), ta cần một cách chuẩn hóa để đảm bảo dữ liệu và kiểu dữ liệu đúng định dạng mong muốn. Đây là lý do chúng ta cần một thư viện cho phép khai báo hoặc định nghĩa kiểu dữ liệu. Có một vài thư viện được tạo ra cho mục đích này, hầu hết đều là mã nguồn mở, và Pydantic là một lựa chọn đã được chấp nhận rộng rãi trong nhiều trường hợp sử dụng.

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 trong Python, được thiết kế để nâng cao độ bền vững và độ tin cậy của mã nguồn. Từ các tác vụ cơ bản như kiểm tra xem một biến có phải là số nguyên hay không, đến các tác vụ phức tạp hơn như đảm bảo các khóa và giá trị trong từ điển lồng nhau có kiểu dữ liệu chính xác, Pydantic có thể xử lý hầu hết mọi kịch bản xác thực dữ liệu với mã nguồn tối thiểu. Trong bài này mình sẽ xác định một số trường hợp phổ biến và đưa ra cách sử dụng Pydantic để giải quyết các trường hợp đó.

Được rồi, bắt đầu thôi!

Type hint trong Python

Trước khi đi vào ví dụ mà mình đã đề cập ở trên, mình sẽ bắt đầu với một số kiến thức cơ bản trong Python.

Qua các phiên bản, Python đã cho ra mắt tính năng Type Hint. Type hint là gì và tại sao chúng ta lại cần nó?

Như ta đã biết thì có rất nhiều ngôn ngữ lập trình như Java, C, C++, PHP, JavaScript, Python,… 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 languages có thể hiểu là khi khai báo hàm và các đối số của nó thì 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). Điều này có nghĩa là khi lập trình, ta phải khai báo kiểu của biến một cách tường minh. Một số ngôn ngữ thuộc kiểu statically typed có thể kể đến như C, C++, Java,…

Dynamically typed languages có thể hiểu là 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). Điều này có nghĩa là khi lập trình chúng ta không cần khai báo kiểu của một biến. Với một biến được khai báo, chúng ta có thể tuỳ ý gán giá trị bất kỳ cho nó mà không cần quan tâm tới kiểu của nó là gì. Nó sẽ tự động ép kiểu khi nhận được giá trị. Một số ngôn ngữ thuộc loại này là Ruby, PHP, Python, JavaScript,…

Có thể thấy Python thuộc loại Dynamically typed languages. Không cần quan tâm kiểu dữ liệu đầu vào, điều này sẽ giúp chúng ta viết code nhanh hơn một chút. Nhưng đi kèm với đó, trong nhiều trường hợp nó sẽ không được tường minh và làm chúng ta gặp khó khăn trong quá trình debug. Một đầu vào tùy biến sẽ làm chúng ta mất nhiều thời gian về những trường hợp của nó có thể xảy ra.

Type hint được ra đời nhằm khắc phục nhược điểm này. 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 để không làm thay đổi bản chất của Dynamically typed languages.

Ví dụ:

def add(a, b): return a + b add(1, 2)
> 3 add(.2, 1)
> 1.2 add('a', 'b')
> 'ab'

Đây là một ví dụ ngắn về cách một hàm đã được định nghĩa có thể được sử dụng trong nhiều trường hợp khác nhau, bao gồm cả những trường hợp mà người viết code không dự đoán trước. Để đảm bảo mã của bạn được sử dụng đúng mục đích, bạn sẽ cần kiểm tra các trường hợp biên và xử lý các trường hợp đó.

Vậy type hint trông như thế nào?

def add(a: int, b: int) -> int: return a + b add(1, 2)
> 3 add(.2, 1)
> 1.2 add('a', 'b')
> 'ab'

Ồ, đoạn code này vẫn hoạt động. Tại sao nhỉ? Bởi vì đây vẫn được gọi là “type hinting” (gợi ý kiểu dữ liệu) chứ không phải là “type enforcing” (ép buộc kiểu dữ liệu). Như đã đề cập trước đó, nó được sử dụng để "thông báo" cho người đọc và "người dùng" về cách sử dụng dự kiến. Một trong những "người dùng" code là các IDE, và IDE bạn chọn nên có khả năng nhận diện và đưa ra cảnh báo nếu bạn cố gắng bỏ qua các khai báo kiểu dữ liệu.

Tại sao chúng ta lại đề cập tới type hint? Bởi vì Pydantic được xây dựng dựa trên tính năng này. Nó sử dụng type hint để định nghĩa các kiểu dữ liệu và cấu trúc dữ liệu, đồng thời thực hiện việc xác thực chúng.

Pydantic - Khởi đầu

Như mình đã đề cập, Pydantic được sử dụng để xác thực các cấu trúc và kiểu của dữ liệu. Có 4 cách mà các bạn có thể sử dụng Pydantic. Trong bài này mình sẽ đưa ra 2 cách sử dụng phổ biến nhất:

• Sử dụng 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)
• Sử dụng BaseModel để định nghĩa và xác thực mô hình dữ liệu thông qua định nghĩa class.

Pydantic - validate_call

Các cụ có câu: “Học đi đôi với hành”. Không có cách nào để tiếp thu kiến thức tốt hơn việc vừa học vừa bắt tay vào thực hành. Do đó chúng ta sẽ cùng học Pydantic thông qua các ví dụ minh hoạ cụ thể.

Trước khi bắt đầu sử dụng Pydantic, bạn sẽ cần cài đặt nó trước. Hãy mở terminal, tạo một môi trường ảo mới và chạy câu lệnh sau để cài đặt Pydantic:

python -m pip install pydantic

Sau đó, bạn cần tạo một project mới, tạo tập lệnh Python đầu tiên, import Pydantic và bắt đầu sử dụng. Ví dụ đầu tiên sẽ là xem lại hàm trước đó của chúng ta và sử dụng Pydantic để đảm bảo nó được sử dụng đúng như mong muốn. Ví dụ:

import pydantic @pydantic.validate_call
def add(a: int, b: int) -> int: return a + b # ---- add(2, 3)
> 5 # ---- add('a', 'a')
> ValidationError: 2 validation errors 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.5/v/int_parsing>
1 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.5/v/int_parsing> # ---- add(.4, .3)
> ValidationError: 2 validation errors for add
0 Input should be a valid integer, got a number with a fractional part [type=int_from_float, input_value=0.4, input_type=float] For further information visit <https://errors.pydantic.dev/2.5/v/int_from_float>
1 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.5/v/int_from_float> # ---- add('3', 'a')
> ValidationError: 1 validation error for add
1 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.5/v/int_parsing> # ---- add('3', '3')
> 6 # ---- add('3', '3.3')
> ValidationError: 1 validation error for add
1 Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='3.3', input_type=str] For further information visit <https://errors.pydantic.dev/2.5/v/int_parsing>

Có một vài điểm ta cần làm rõ:

• validate_call được sử dụng như một decorator. Về cơ bản, đây là một lớp bao quanh hàm được khai báo, giới thiệu thêm logic có thể chạy khi hàm được định nghĩa cũng như khi bạn gọi hàm. Ở đây, nó đượ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).
• Một lần gọi hàm đã được xác thực sẽ phát sinh ValidationError trong trường hợp bạn sử dụng hàm theo cách không mong muốn. Lỗi này được chú thích chi tiết và giải thích rõ lý do tại sao nó phát sinh.
• Theo nguyên tắc “khoan dung”, 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.

Vậy còn các giá trị mặc định và trích xuất đối số thì sao?

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)
> ValidationError: 1 validation error for add
a Missing required keyword only argument [type=missing_keyword_only_argument, input_value=ArgsKwargs((4, 3, 4)), input_type=ArgsKwargs] For further information visit <https://errors.pydantic.dev/2.5/v/missing_keyword_only_argument> # ---- add(4, 3, 4, a=3)
> 18 # ---- @validate_call
def add(*args: int, a: int, b: int = 4) -> int: return str(sum(args) + a + b) # ----  add(4, 3, 4, a=3)
> '18'

Một số điều bạn có thể rút ra từ ví dụ này là:

• Bạn có thể chú thích kiểu của khai báo số lượng đối số biến thiên (*args).
• Các giá trị mặc định vẫn là tùy chọn, ngay cả khi bạn đang chú thích các kiểu dữ liệu biến.
• validate_call chấp nhận đối số validate_return, điều này giúp xác thực giá trị trả về của hàm. Chú ý rằng ép kiểu dữ liệu cũng được áp dụng trong trường hợp này. validate_return được đặt mặc định là False. Nếu để nguyên như vậy, hàm có thể không trả về giá trị như đã khai báo trong type hint.

Nếu bạn muốn xác thực kiểu dữ liệu nhưng cũng giới hạn các giá trị mà biến có thể nhận, ví dụ:

from pydantic import validate_call, Field
from typing import Annotated type_age = Annotated[int, Field(lt=120)] @validate_call(validate_return=True)
def add(age_one: int, age_two: type_age) -> int: return age_one + age_two add(3, 300)
> ValidationError: 1 validation error for add
1 Input should be less than 120 [type=less_than, input_value=200, input_type=int] For further information visit <https://errors.pydantic.dev/2.5/v/less_than>

Ví dụ này chỉ ra rằng:

• Bạn có thể sử dụng Annotated và pydantic.Field không chỉ để xác thực kiểu dữ liệu mà còn để thêm metadata mà Pydantic sử dụng để ràng buộc các giá trị và định dạng của biến.
• ValidationError lại một lần nữa cung cấp thông tin chi tiết về vấn đề trong lần gọi hàm của chúng ta. Điều này rất hữu ích khi bạn cần debug trong tương lai.

Sau đây là một ví dụ nữa về cách bạn có thể vừa xác thực vừa ràng buộc các giá trị của biến. Chúng ta sẽ mô phỏng một payload (từ điển) mà bạn muốn xử lý trong hàm của mình sau khi nó đã được xác thực.

from pydantic import HttpUrl, PastDate
from pydantic import Field
from pydantic import validate_call
from typing import Annotated Name = Annotated[str, Field(min_length=2, max_length=15)] @validate_call(validate_return=True)
def process_payload(url: HttpUrl, name: Name, birth_date: PastDate) -> str: return f'{name=}, {birth_date=}' # ---- payload = { 'url': 'httpss://example.com', 'name': 'A', 'birth_date': '2024-12-12'
} process_payload(**payload)
> ValidationError: 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.5/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.5/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.5/v/date_past> # ---- payload = { 'url': '<https://example.com>', 'name': 'Alex-1234567891011121314', 'birth_date': '2020-12-12'
} process_payload(**payload)
> ValidationError: 1 validation error for process_payload
name String should have at most 15 characters [type=string_too_long, input_value='Alex-1234567891011121314', input_type=str] For further information visit <https://errors.pydantic.dev/2.5/v/string_too_long>

Đây là những kiến thức cơ bản về cách xác thực các đối số hàm và giá trị trả về của chúng.

Bây giờ, chúng ta sẽ chuyển sang cách quan trọng thứ hai mà Pydantic có thể được sử dụng để xác thực và xử lý dữ liệu: thông qua việc định nghĩa các mô hình.

Pydantic - BaseModel

Như các bạn sẽ thấy, phần này sẽ thú vị hơn cho mục đích xử lý dữ liệu.

Cho đến thời điểm hiện tại, chúng ta đã sử dụng validate_call để “trang trí” (decorate) các hàm và chỉ định các đối số hàm cùng với các kiểu và ràng buộc tương ứng.

Ở đây, chúng ta định nghĩa các mô hình bằng cách khai báo các lớp mô hình (model class), nơi chúng ta chỉ định các trường, kiểu dữ liệu của chúng, và các ràng buộc. Điều này tương tự với những gì ta đã làm trước đó. Bằng cách định nghĩa một lớp mô hình kế thừa từ Pydantic 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, 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.

Dưới đây là một ví dụ:

from pydantic import Field
from pydantic import BaseModel class Person(BaseModel): name: str = Field(min_length=2, max_length=15) age: int = Field(gt=0, lt=120) # ---- john = Person(name='john', age=20)
> Person(name='john', age=20) # ---- mike = Person(name='m', age=0)
> ValidationError: 2 validation errors for Person
name String should have at least 2 characters [type=string_too_short, input_value='j', input_type=str] For further information visit <https://errors.pydantic.dev/2.5/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.5/v/greater_than>

Bạn có thể sử dụng chú thích (annotation) ở đây, và cũng có thể chỉ định các giá trị mặc định cho các trường. Hãy xem xét một ví dụ khác:

from pydantic import Field
from pydantic import BaseModel
from typing import Annotated Name = Annotated[str, Field(min_length=2, max_length=15)]
Age = Annotated[int, Field(default=1, ge=0, le=120)] class Person(BaseModel): name: Name age: Age # ---- mike = Person(name='mike')
> Person(name='mike', age=1)

Mọi thứ trở nên rất thú vị khi use case của bạn trở nên phức tạp hơn. Bạn còn nhớ payload mà ta đã định nghĩa ở trên chứ? Mình sẽ định nghĩa một cấu trúc phức tạp hơn mà ta sẽ đi qua và xác thực nó. Ở đây, ta sẽ tạo một payload mà ta có thể sử dụng để truy vấn một dịch vụ hoạt động như một trung gian giữa chúng ta và các nhà cung cấp mô hình ngôn ngữ lớn (LLM). Sau đó, chúng ta sẽ xác thực nó.

Dưới đây là một ví dụ:

from pydantic import Field
from pydantic import BaseModel
from pydantic import ConfigDict from typing import Literal
from typing import Annotated
from enum import Enum payload = { "req_id": "test", "text": "This is a sample text.", "instruction": "embed", "llm_provider": "openai", "llm_params": { "llm_temperature": 0, "llm_model_name": "gpt4o" }, "misc": "what"
} ReqID = Annotated[str, Field(min_length=2, max_length=15)] 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) text: str = Field(min_length=5) instruction: Literal['embed', 'chat'] llm_provider: LLMProviders llm_params: LLMParams # model_config = ConfigDict(use_enum_values=True) # ---- validated_payload = Payload(**payload)
validated_payload
> 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'}} # ---- # Sau khi them doan code
# model_config = ConfigDict(use_enum_values=True)
# trong phan dinh nghia Payload model, ban se nhan duoc: validated_payload.model_dump(by_alias=True)
> {'text': 'This is a sample text.', 'instruction': 'embed', 'llm_provider': 'openai', 'llm_params': {'temperature': 0, 'model': 'gpt4o'}}

Một số điểm đáng chú ý từ ví dụ này là:

• Bạn có thể sử dụng Enum hoặc Literal để định nghĩa danh sách các giá trị cụ thể được mong đợi.
• Trong trường hợp bạn muốn đặt tên cho trường của mô hình khác với tên trường trong dữ liệu được xác thực, bạn có thể sử dụng validation_alias. Nó chỉ định tên trường trong dữ liệu đang được xác thực.
• serialization_alias được sử dụng khi tên trường nội bộ của mô hình 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.
• Trường có thể được loại trừ khỏi quá trình tuần tự hóa với exclude=True.
• Các trường mô hình có thể là các mô hình Pydantic. Quá trình xác thực trong trường hợp này được thực hiện đệ quy. Điều này sẽ rất hữu ích vì Pydantic sẽ thực hiện việc đi sâu vào bên trong trong khi xác thực các cấu trúc lồng nhau.
• Các trường không được tính đến trong định nghĩa mô hình sẽ không được phân tích.

Pydantic - Use Cases

Ở đây mình sẽ trình bày các đoạn code để cho thấy bạn có thể sử dụng Pydantic như thế nào trong các tác vụ hàng ngày của mình.

Xử lý dữ liệu

Giả sử bạn có dữ liệu cần xác thực và xử lý. Dữ liệu của bạn có thể được lưu trữ trong file CSV, các file Parquet hoặc có thể là một cơ sở dữ liệu NoSQL dưới dạng một tài liệu. Hãy lấy ví dụ về một file CSV và giả sử bạn muốn xử lý nội dung của nó.

Đây là ví dụ về file CSV (test.csv):

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

Và đây là cách để xác thực và phân tích cú pháp dữ liệu:

from pydantic import BaseModel
from pydantic import Field from pydantic import field_validator
from pydantic import ValidationInfo
from typing import List
import csv FILE_NAME = 'test.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() class ValidatedModels(BaseModel): validated: List[DataModel] validated_rows = [] with open(FILE_NAME, 'r') as f: reader = csv.DictReader(f, delimiter=',') for row in reader: try: validated_rows.append(DataModel(**row)) except ValidationError as ve: # print out error # disregard the record print(f'{ve=}') validated_rows
> [DataModel(name='Matt', age=10, bank_account=0.0), DataModel(name='Abraham', age=100, bank_account=100000.0), DataModel(name='Mary', age=15, bank_account=15.0)] validated = ValidatedModels(validated=validated_rows)
validated.model_dump()
> {'validated': [{'name': 'Matt', 'age': 10, 'bank_account': 0.0}, {'name': 'Abraham', 'age': 100, 'bank_account': 100000.0}, {'name': 'Mary', 'age': 15, 'bank_account': 15.0}]}

Xác thực FastAPI request

FastAPI đã được tích hợp với Pydantic, vì vậy phần này mình sẽ nói rất ngắn gọn. 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 mà chúng ta đã đề cập ở đầu bài viết này.

Dưới đây là ví dụ về app.py được sử dụng để chạy dịch vụ dựa trên FastAPI:

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

Kết luận

Pydantic là một thư viện mạnh mẽ và có nhiều cơ chế cho rất nhiều trường hợp sử dụng cũng như các trường hợp đặc biệt khác nhau. Trong bài này, mình đã giải thích cơ bản về cách sử dụng Pydantic. Dưới đây mình sẽ đưa ra các tài liệu tham khảo cho những ai muốn tìm hiểu kĩ hơn về thư viện này.

Tài liệu tham khảo

• Python type hinting
• Pydantic models
• Validate call decorator
• Pydantic fields
• Pydantic validators
• Real Python's Tutorial