Protocols — Duck Typing có kiểm tra kiểu tĩnh Nâng cao

Python nổi tiếng với duck typing — "nếu nó kêu quạc quạc và đi lạch bạch, đó là con vịt". Triết lý này cho phép viết code linh hoạt đến kinh ngạc: bất kỳ object nào có đủ method cần thiết đều dùng được, không cần kế thừa từ base class nào cả. Nhưng khi codebase có 200 engineer contribute hàng ngày, duck typing trở thành bug typing. Một developer đổi tên method từ process sang execute, mười module khác vỡ mà không ai biết cho đến khi production sập lúc nửa đêm.

PEP 544 giới thiệu typing.Protocol — câu trả lời của Python cho bài toán kinh điển: làm sao giữ sự linh hoạt của duck typing mà vẫn có type checker bắt lỗi trước khi code chạy? Protocol mang đến structural subtyping — một khái niệm lấy cảm hứng trực tiếp từ interface của Go. Thay vì yêu cầu class phải khai báo "tôi implement interface X" (nominal typing), Protocol chỉ kiểm tra: "class này có đủ method và attribute cần thiết không?" Nếu có — tương thích. Không cần register(), không cần kế thừa, không cần biết Protocol tồn tại.

Đây không phải chỉ là syntactic sugar cho type hints. Protocol thay đổi cách bạn thiết kế architecture: plugin systems không cần base class, API contracts được enforce tại compile time, dependency injection trở nên tự nhiên như viết function bình thường. Sau bài này, bạn sẽ nhìn duck typing bằng con mắt hoàn toàn khác — có hệ thống, có kiểm soát, sẵn sàng cho production.

Bức tranh tư duy

Ổ cắm điện — Protocol trong đời thực

Hãy nghĩ về ổ cắm điện ba chấu tiêu chuẩn trong nhà bạn. Ổ cắm không quan tâm thiết bị nào cắm vào — quạt, tủ lạnh, máy tính, hay bất kỳ thiết bị nào từ bất kỳ nhà sản xuất nào. Điều kiện duy nhất: phích cắm phải có đúng hình dạng — ba chấu, đúng khoảng cách, đúng kích thước. Không cần "đăng ký" với ổ cắm, không cần giấy chứng nhận "tôi tương thích với ổ cắm loại A". Chỉ cần hình dạng khớp — dùng được.

Protocol trong Python hoạt động y hệt. Khi bạn định nghĩa một Protocol với method process() và validate(), bất kỳ class nào có hai method đó — dù được viết ở đâu, bởi ai, khi nào — đều tự động tương thích. Type checker (mypy, pyright) sẽ kiểm tra "hình dạng" tại thời điểm phân tích code, không phải lúc runtime.

So sánh với ABC — cách tiếp cận nominal typing: ABC giống như ổ cắm yêu cầu bạn phải mang thiết bị đến trung tâm kiểm định, điền đơn đăng ký (class MyDevice(PowerSocket):), rồi mới được cắm. Hiệu quả? Có. Nhưng mất đi sự linh hoạt khi bạn muốn dùng thiết bị từ nhà sản xuất không biết đến tiêu chuẩn của bạn.

ABC (Nominal Typing):          Protocol (Structural Typing):
┌──────────────────┐            ┌──────────────────┐
│   PowerSocket    │            │   PowerSocket     │
│   (ABC)          │            │   (Protocol)      │
│                  │            │                   │
│  + connect()     │            │  + connect()      │
│  + disconnect()  │            │  + disconnect()   │
└────────┬─────────┘            └──────────────────┘
         │ inherit                       ▲
         │ (bắt buộc)                    │ structural match
┌────────┴─────────┐            ┌────────┴─────────┐
│    MyDevice      │            │    MyDevice       │
│ (PowerSocket)    │            │ (standalone)      │
│                  │            │                   │
│  + connect()     │            │  + connect()      │
│  + disconnect()  │            │  + disconnect()   │
└──────────────────┘            └──────────────────┘
  "Tôi ĐĂNG KÝ là                "Tôi CÓ đủ method,
   PowerSocket"                    tự động tương thích"

Cốt lõi kỹ thuật

Structural vs Nominal Typing — hai triết lý

Nominal typing (ABC, Java interface): kiểu được xác định bởi tên và quan hệ kế thừa. Dù hai class có method giống hệt nhau, chúng không tương thích trừ khi cùng kế thừa từ base class.

python

from abc import ABC, abstractmethod

class Drawable(ABC):
    @abstractmethod
    def draw(self) -> None: ...

class Circle(Drawable):        # ✅ Tương thích — kế thừa Drawable
    def draw(self) -> None:
        print("Drawing circle")

class Square:                   # ❌ KHÔNG tương thích — dù có draw()
    def draw(self) -> None:
        print("Drawing square")

def render(shape: Drawable) -> None:
    shape.draw()

render(Circle())   # ✅ OK
render(Square())   # ❌ mypy error: Square không phải Drawable

Structural typing (Protocol): kiểu được xác định bởi cấu trúc — tập hợp method và attribute. Nếu class có đủ "hình dạng", nó tương thích.

python

from typing import Protocol

class Drawable(Protocol):
    def draw(self) -> None: ...

class Circle:                   # ✅ Tương thích — có draw()
    def draw(self) -> None:
        print("Drawing circle")

class Square:                   # ✅ Tương thích — cũng có draw()
    def draw(self) -> None:
        print("Drawing square")

def render(shape: Drawable) -> None:
    shape.draw()

render(Circle())   # ✅ OK
render(Square())   # ✅ OK — không cần kế thừa gì cả

Sự khác biệt là cốt lõi: Circle và Square không biết Drawable tồn tại, nhưng type checker vẫn xác nhận chúng tương thích.

typing.Protocol — định nghĩa và sử dụng

Protocol được định nghĩa bằng cách kế thừa typing.Protocol. Mỗi method và attribute trong Protocol body trở thành yêu cầu structural mà class phải thỏa mãn.

python

from typing import Protocol, runtime_checkable

class Closeable(Protocol):
    """Bất kỳ object nào có thể đóng."""
    def close(self) -> None: ...

class Readable(Protocol):
    """Bất kỳ object nào có thể đọc data."""
    def read(self, size: int = -1) -> bytes: ...
    def close(self) -> None: ...

# Kết hợp nhiều Protocol — intersection
class ReadWritable(Protocol):
    def read(self, size: int = -1) -> bytes: ...
    def write(self, data: bytes) -> int: ...
    def close(self) -> None: ...

Sử dụng Protocol trong type hints giống y hệt ABC — nhưng class implement không cần biết Protocol tồn tại:

python

import socket

class DatabaseConnection:
    """Class này không import hay kế thừa Closeable."""
    def __init__(self, host: str, port: int) -> None:
        self._sock = socket.socket()
        self._sock.connect((host, port))

    def close(self) -> None:
        self._sock.close()

def cleanup(resource: Closeable) -> None:
    """Nhận bất kỳ object nào có method close()."""
    resource.close()
    print("Resource cleaned up")

conn = DatabaseConnection("localhost", 5432)
cleanup(conn)  # ✅ mypy happy — DatabaseConnection thỏa mãn Closeable

@runtime_checkable và isinstance()

Mặc định, Protocol chỉ hoạt động tại type-check time (khi chạy mypy/pyright). Để dùng isinstance() tại runtime, thêm decorator @runtime_checkable:

python

from typing import Protocol, runtime_checkable

@runtime_checkable
class Renderable(Protocol):
    def render(self) -> str: ...

class HTMLWidget:
    def render(self) -> str:
        return "<div>Widget</div>"

class PlainText:
    def render(self) -> str:
        return "Plain text content"

class NotRenderable:
    def display(self) -> str:
        return "I don't have render()"

# isinstance() hoạt động tại runtime
widget = HTMLWidget()
print(isinstance(widget, Renderable))       # True
print(isinstance(NotRenderable(), Renderable))  # False

# Sử dụng trong logic runtime
def safe_render(obj: object) -> str:
    if isinstance(obj, Renderable):
        return obj.render()
    return str(obj)

Cảnh báo quan trọng

@runtime_checkable chỉ kiểm tra sự tồn tại của method, không kiểm tra signature (tham số, return type). Một class có render(self, x: int) vẫn pass isinstance() check dù signature không khớp. Type checker mới bắt được lỗi signature.

Protocol với attributes, properties và class variables

Protocol không chỉ định nghĩa method — còn hỗ trợ attributes, properties, và class variables:

python

from typing import Protocol, ClassVar, runtime_checkable

@runtime_checkable
class Configurable(Protocol):
    name: str                                   # Instance attribute
    version: ClassVar[str]                      # Class variable
    @property
    def is_active(self) -> bool: ...            # Property
    def configure(self, **kwargs: str) -> None: ...  # Method


class RedisCache:
    version: ClassVar[str] = "2.1.0"

    def __init__(self, name: str) -> None:
        self.name = name
        self._active = True

    @property
    def is_active(self) -> bool:
        return self._active

    def configure(self, **kwargs: str) -> None:
        for key, value in kwargs.items():
            setattr(self, f"_{key}", value)


def initialize(component: Configurable) -> None:
    print(f"Initializing {component.name} v{type(component).version}")
    if component.is_active:
        component.configure(timeout="30", retry="3")

initialize(RedisCache("main-cache"))  # ✅ Structural match

Generic Protocols — Protocol[T]

Protocol kết hợp với generics tạo ra abstraction cực mạnh — type-safe mà vẫn linh hoạt:

python

from typing import Protocol, TypeVar, Sequence

T = TypeVar("T")
T_co = TypeVar("T_co", covariant=True)
T_contra = TypeVar("T_contra", contravariant=True)

class Repository(Protocol[T]):
    """Generic repository — hoạt động với bất kỳ entity type nào."""
    def get(self, id: str) -> T | None: ...
    def save(self, entity: T) -> None: ...
    def delete(self, id: str) -> bool: ...


class User:
    def __init__(self, id: str, email: str) -> None:
        self.id = id
        self.email = email


class InMemoryUserRepo:
    """Không kế thừa Repository — chỉ cần có đúng methods."""
    def __init__(self) -> None:
        self._store: dict[str, User] = {}

    def get(self, id: str) -> User | None:
        return self._store.get(id)

    def save(self, entity: User) -> None:
        self._store[entity.id] = entity

    def delete(self, id: str) -> bool:
        return self._store.pop(id, None) is not None


def find_or_fail(repo: Repository[T], id: str) -> T:
    """Hoạt động với BẤT KỲ Repository nào."""
    entity = repo.get(id)
    if entity is None:
        raise KeyError(f"Not found: {id}")
    return entity

user_repo = InMemoryUserRepo()
user_repo.save(User("1", "alice@example.com"))
print(find_or_fail(user_repo, "1"))  # ✅ mypy infers T = User

Covariant và Contravariant Protocol — khi cần kiểm soát chiều biến thiên:

python

class Serializer(Protocol[T_contra]):
    """Contravariant: Serializer[Animal] chấp nhận ở nơi cần Serializer[Dog]."""
    def serialize(self, obj: T_contra) -> bytes: ...

class Producer(Protocol[T_co]):
    """Covariant: Producer[Dog] chấp nhận ở nơi cần Producer[Animal]."""
    def produce(self) -> T_co: ...

Thực chiến

Xây dựng Plugin System cho Data Processing Framework

Bài toán thực tế: bạn đang xây dựng framework xử lý dữ liệu. Các team khác nhau viết plugin xử lý từng loại data (CSV, JSON, Parquet...). Yêu cầu: plugin phải type-safe, nhưng không bắt buộc import hay kế thừa bất cứ thứ gì từ framework.

python

"""
data_framework/protocols.py
Định nghĩa contracts cho plugin system.
"""
from __future__ import annotations

from typing import Protocol, Any, runtime_checkable, TypeVar, ClassVar
from dataclasses import dataclass


@dataclass(frozen=True)
class Schema:
    """Mô tả cấu trúc dữ liệu."""
    fields: dict[str, type]
    required: frozenset[str]
    name: str


@runtime_checkable
class DataProcessor(Protocol):
    """Contract cho mọi data processor plugin.

    Bất kỳ class nào có đủ 3 methods dưới đây
    đều là DataProcessor hợp lệ — không cần kế thừa.
    """
    format_name: ClassVar[str]

    def validate(self, raw_data: bytes) -> bool:
        """Kiểm tra data có hợp lệ cho format này không."""
        ...

    def process(self, raw_data: bytes) -> list[dict[str, Any]]:
        """Parse raw bytes thành list of records."""
        ...

    def get_schema(self) -> Schema:
        """Trả về schema mô tả output structure."""
        ...

Các team viết plugin độc lập, không import Protocol:

python

"""
plugins/csv_processor.py — Không import DataProcessor Protocol.
"""
import csv
import io
from typing import Any, ClassVar

from data_framework.protocols import Schema


class CsvProcessor:
    format_name: ClassVar[str] = "csv"

    def __init__(self, delimiter: str = ",", encoding: str = "utf-8") -> None:
        self._delimiter = delimiter
        self._encoding = encoding

    def validate(self, raw_data: bytes) -> bool:
        try:
            text = raw_data.decode(self._encoding)
            header = next(csv.reader(io.StringIO(text), delimiter=self._delimiter))
            return len(header) > 0
        except (UnicodeDecodeError, StopIteration, csv.Error):
            return False

    def process(self, raw_data: bytes) -> list[dict[str, Any]]:
        text = raw_data.decode(self._encoding)
        return list(csv.DictReader(io.StringIO(text), delimiter=self._delimiter))

    def get_schema(self) -> Schema:
        return Schema(fields={}, required=frozenset(), name="csv_output")

Framework discover và validate plugins tại runtime:

python

"""
data_framework/registry.py — Plugin registry với structural typing.
"""
from __future__ import annotations

import logging
from typing import Any

from data_framework.protocols import DataProcessor

logger = logging.getLogger(__name__)


class PluginRegistry:
    """Type-safe plugin registry sử dụng Protocol."""

    def __init__(self) -> None:
        self._processors: dict[str, DataProcessor] = {}

    def register(self, processor: DataProcessor) -> None:
        """Đăng ký processor — mypy kiểm tra structural compatibility."""
        format_name = type(processor).format_name
        if format_name in self._processors:
            raise ValueError(f"Processor for '{format_name}' already registered")
        self._processors[format_name] = processor
        logger.info("Registered processor: %s", format_name)

    def process_data(self, format_name: str, raw_data: bytes) -> list[dict[str, Any]]:
        processor = self._processors.get(format_name)
        if processor is None:
            raise KeyError(f"No processor for format: {format_name}")
        if not processor.validate(raw_data):
            raise ValueError(f"Invalid {format_name} data")
        return processor.process(raw_data)


# --- Sử dụng ---
registry = PluginRegistry()
registry.register(CsvProcessor(delimiter=","))

csv_data = b"name,age\nAlice,30\nBob,25"
records = registry.process_data("csv", csv_data)
for record in records:
    print(record)
# {'name': 'Alice', 'age': '30'}
# {'name': 'Bob', 'age': '25'}

Kiến trúc này cho phép bất kỳ ai viết processor mới mà không cần import base class từ framework — chỉ cần đúng "hình dạng".

Sai lầm điển hình

❌ SAI #1: Kế thừa Protocol khi không cần thiết

python

# ❌ SAI — Mất đi ý nghĩa của structural subtyping
from typing import Protocol

class Loggable(Protocol):
    def log(self, message: str) -> None: ...

class FileLogger(Loggable):  # ← Không cần kế thừa!
    def log(self, message: str) -> None:
        with open("app.log", "a") as f:
            f.write(message + "\n")

python

# ✅ ĐÚNG — Standalone class, structural match tự động
from typing import Protocol

class Loggable(Protocol):
    def log(self, message: str) -> None: ...

class FileLogger:  # Không kế thừa gì cả
    def log(self, message: str) -> None:
        with open("app.log", "a") as f:
            f.write(message + "\n")

def setup_logging(logger: Loggable) -> None:
    logger.log("System initialized")

setup_logging(FileLogger())  # ✅ mypy: OK — structural match

Production impact: Kế thừa Protocol tạo coupling không cần thiết. Khi Protocol thay đổi, tất cả class kế thừa phải update — đánh mất lợi thế decoupling của structural typing.

❌ SAI #2: Tin tưởng runtime_checkable kiểm tra signature

python

# ❌ SAI — isinstance() KHÔNG kiểm tra parameter types
from typing import Protocol, runtime_checkable

@runtime_checkable
class Transformer(Protocol):
    def transform(self, data: list[float]) -> list[float]: ...

class BrokenTransformer:
    def transform(self) -> str:  # Signature hoàn toàn sai!
        return "not a list"

obj = BrokenTransformer()
print(isinstance(obj, Transformer))  # True! 💥 Sai nhưng pass
# obj.transform(data=[1.0, 2.0])    # TypeError tại runtime

python

# ✅ ĐÚNG — Kết hợp isinstance() với validation bổ sung
import inspect

def is_valid_transformer(obj: object) -> bool:
    """Kiểm tra sự tồn tại VÀ signature cơ bản."""
    if not isinstance(obj, Transformer):
        return False
    sig = inspect.signature(getattr(obj, "transform"))
    params = [p for p in sig.parameters.values() if p.name != "self"]
    return len(params) >= 1  # Ít nhất 1 parameter ngoài self

Production impact: Code tin tưởng isinstance() hoàn toàn sẽ crash tại runtime khi plugin có method cùng tên nhưng signature khác. Luôn kết hợp với type checker hoặc validation bổ sung.

❌ SAI #3: Fat Protocol — vi phạm Interface Segregation

python

# ❌ SAI — Protocol quá lớn, buộc implement nhiều thứ không cần
from typing import Protocol, Any

class DataHandler(Protocol):
    def read(self, source: str) -> bytes: ...
    def write(self, dest: str, data: bytes) -> None: ...
    def transform(self, data: bytes) -> bytes: ...
    def validate(self, data: bytes) -> bool: ...
    def compress(self, data: bytes) -> bytes: ...
    def encrypt(self, data: bytes) -> bytes: ...
    def log(self, message: str) -> None: ...
    def notify(self, event: str) -> None: ...

python

# ✅ ĐÚNG — Tách thành nhiều Protocol nhỏ, composable
from typing import Protocol

class Readable(Protocol):
    def read(self, source: str) -> bytes: ...

class Writable(Protocol):
    def write(self, dest: str, data: bytes) -> None: ...

class Transformable(Protocol):
    def transform(self, data: bytes) -> bytes: ...

class Validatable(Protocol):
    def validate(self, data: bytes) -> bool: ...

# Kết hợp khi cần — class chỉ implement những gì nó cần
def etl_pipeline(
    reader: Readable,
    transformer: Transformable,
    writer: Writable,
) -> None:
    raw = reader.read("input.csv")
    processed = transformer.transform(raw)
    writer.write("output.parquet", processed)

Production impact: Fat Protocol buộc mọi implementation phải có đầy đủ 8 methods, dù chỉ cần 2. Kết quả: code stub vô nghĩa, class phình to, khó maintain.

❌ SAI #4: Nhầm Protocol với ABC — dùng sai tool

python

# ❌ SAI — Dùng Protocol khi cần shared implementation
from typing import Protocol

class Cache(Protocol):
    def get(self, key: str) -> bytes | None: ...
    def set(self, key: str, value: bytes, ttl: int = 300) -> None: ...
    # Muốn method chung cho mọi cache... nhưng Protocol không cho phép!
    # def get_or_set(self, key, factory, ttl): ← Không nên đặt logic ở đây

python

# ✅ ĐÚNG — ABC khi cần shared implementation, Protocol khi cần structural contract
from abc import ABC, abstractmethod
from typing import Protocol, Callable

# Protocol cho external/third-party code — structural matching
class CacheReader(Protocol):
    def get(self, key: str) -> bytes | None: ...

# ABC cho internal hierarchy — shared logic
class BaseCache(ABC):
    @abstractmethod
    def get(self, key: str) -> bytes | None: ...

    @abstractmethod
    def set(self, key: str, value: bytes, ttl: int = 300) -> None: ...

    def get_or_set(
        self,
        key: str,
        factory: Callable[[], bytes],
        ttl: int = 300,
    ) -> bytes:
        """Shared logic — lý do chính để dùng ABC."""
        value = self.get(key)
        if value is None:
            value = factory()
            self.set(key, value, ttl)
        return value

Production impact: Protocol không hỗ trợ default implementation. Nếu bạn cần shared logic giữa các subclass, ABC là lựa chọn đúng. Protocol dành cho contracts giữa các module/team, không phải class hierarchy nội bộ.

Under the Hood

mypy giải quyết structural subtyping như thế nào

Khi mypy gặp một hàm nhận parameter kiểu Protocol, nó thực hiện structural compatibility check — không dựa vào MRO (Method Resolution Order) hay __bases__, mà so sánh từng member của Protocol với class:

mypy check: "CsvProcessor thỏa mãn DataProcessor?"

DataProcessor.validate(bytes) -> bool    → CsvProcessor.validate(bytes) -> bool     ✅
DataProcessor.process(bytes) -> list     → CsvProcessor.process(bytes) -> list      ✅
DataProcessor.get_schema() -> Schema     → CsvProcessor.get_schema() -> Schema      ✅
DataProcessor.format_name: ClassVar[str] → CsvProcessor.format_name = "csv"         ✅
→ Kết luận: CsvProcessor tương thích.

Quá trình này kiểm tra:

Method existence: class có method cùng tên không?
Parameter compatibility: signature có tương thích không? (contravariant cho parameters)
Return type compatibility: return type có tương thích không? (covariant cho return)
Attribute types: attribute có đúng kiểu không?

runtime_checkable và ABCMeta.subclasshook

Bên dưới, @runtime_checkable tận dụng cơ chế __subclasshook__ của ABCMeta. Khi bạn gọi isinstance(obj, SomeProtocol):

python

# CPython simplified — typing.py
class _ProtocolMeta(ABCMeta):
    def __instancecheck__(cls, instance):
        if not cls._is_runtime_checkable:
            raise TypeError(
                "Protocols with non-method members don't support "
                "isinstance()"
            )
        # Chỉ kiểm tra SỰ TỒN TẠI, không kiểm tra signature
        for attr in cls.__protocol_attrs__:
            if not hasattr(instance, attr):
                return False
        return True

Luồng thực thi cụ thể:

isinstance(obj, Renderable) gọi Renderable.__instancecheck__(obj)
__instancecheck__ lấy __protocol_attrs__ — set tên các method/attribute
Với mỗi attr, gọi hasattr(obj, attr) — chỉ kiểm tra tên, không kiểm tra type
Nếu tất cả tồn tại → True, thiếu một → False

Performance implications

isinstance() với Protocol chậm hơn ABC 3-5x vì phải iterate qua __protocol_attrs__ và gọi hasattr() cho mỗi attribute, trong khi ABC chỉ check __mro__ (O(1) với cache).

Check type	Thời gian ước lượng
Regular class `isinstance()`	~0.05 μs/call
ABC `isinstance()`	~0.2 μs/call
Protocol `isinstance()`	~0.8 μs/call

Quy tắc thực hành: Tránh isinstance() với Protocol trong hot path. Dùng type checker tại build time, chỉ dùng runtime check ở entry points (plugin loading, API boundary).

Protocol tại runtime vs type-check time

Sự khác biệt quan trọng mà nhiều developer bỏ qua:

Hành vi	Type-check time (mypy)	Runtime
Kiểm tra method tồn tại	✅ Có	✅ Có (nếu `@runtime_checkable`)
Kiểm tra parameter types	✅ Có	❌ Không
Kiểm tra return types	✅ Có	❌ Không
Kiểm tra attribute types	✅ Có	❌ Không (chỉ tồn tại)
Kiểm tra property vs method	✅ Có	❌ Không
Performance impact	Zero (build step)	Có (mỗi isinstance call)

Kết luận: type checker là nguồn kiểm tra chính. @runtime_checkable chỉ là lớp phòng thủ bổ sung cho dynamic code (plugin loading, deserialization).

Checklist ghi nhớ

✅ Checklist triển khai

Protocol fundamentals

[ ] Phân biệt được structural typing (Protocol) và nominal typing (ABC) — biết khi nào dùng cái nào
[ ] Định nghĩa Protocol với method stubs sử dụng ... (ellipsis), không cần implementation
[ ] Hiểu rằng class implement Protocol không cần kế thừa Protocol — structural match tự động
[ ] Sử dụng @runtime_checkable khi cần isinstance() check tại runtime

Design patterns

[ ] Áp dụng Interface Segregation — tách Protocol nhỏ, composable thay vì fat Protocol
[ ] Dùng Protocol cho cross-module contracts, ABC cho internal hierarchy với shared logic
[ ] Kết hợp Generic Protocol (Protocol[T]) cho type-safe abstractions
[ ] Hiểu covariant (T_co) và contravariant (T_contra) trong Generic Protocol

Production practices

[ ] Đặt Protocol definitions trong module riêng (protocols.py) — tránh circular imports
[ ] Luôn chạy mypy/pyright trong CI pipeline — Protocol chỉ phát huy tối đa với type checker
[ ] Không dùng isinstance() với Protocol trong hot path — chậm hơn ABC 3-5x
[ ] Kết hợp runtime_checkable với validation bổ sung — isinstance() không kiểm tra signatures

Pitfalls

[ ] Không kế thừa Protocol trong implementation class — tạo coupling không cần thiết
[ ] Không tin tưởng runtime_checkable 100% — nó chỉ check method names, không check types
[ ] Kiểm tra Protocol compatibility trong CI, không phải chỉ local — đảm bảo mọi team tuân thủ

Bài tập luyện tập

Bài 1: Serializable Protocol (Intermediate)

Yêu cầu: Định nghĩa Protocol Serializable với hai methods:

to_dict(self) -> dict[str, Any] — chuyển object thành dictionary
from_dict(cls, data: dict[str, Any]) -> Self — class method tạo object từ dictionary

Implement 3 class: UserProfile, ProductListing, OrderRecord — mỗi class có cấu trúc data khác nhau, không kế thừa Serializable. Viết hàm serialize_batch() nhận list[Serializable] và trả về JSON string.

🧠 Quiz

Câu hỏi kiểm tra: Khi nào nên dùng Protocol thay vì ABC?

[ ] A. Khi muốn enforce implementation bắt buộc tại runtime
[ ] B. Khi cần shared default methods giữa các subclass
[x] C. Khi muốn type-safe contracts mà không tạo coupling qua inheritance
[ ] D. Khi cần mixin functionality Giải thích: Protocol dùng structural subtyping — class không cần biết Protocol tồn tại mà vẫn tương thích. Điều này loại bỏ coupling qua inheritance, lý tưởng cho cross-module và cross-team contracts. ABC phù hợp hơn cho trường hợp A và B vì nó hỗ trợ abstractmethod enforcement và default implementations.

Xem lời giải Bài 1

python

from __future__ import annotations

import json
from typing import Any, Protocol, Self


class Serializable(Protocol):
    def to_dict(self) -> dict[str, Any]: ...

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Self: ...


class UserProfile:
    def __init__(self, user_id: str, email: str, tier: str) -> None:
        self.user_id = user_id
        self.email = email
        self.tier = tier

    def to_dict(self) -> dict[str, Any]:
        return {"user_id": self.user_id, "email": self.email, "tier": self.tier}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Self:
        return cls(user_id=data["user_id"], email=data["email"], tier=data["tier"])


class ProductListing:
    def __init__(self, sku: str, name: str, price: float) -> None:
        self.sku = sku
        self.name = name
        self.price = price

    def to_dict(self) -> dict[str, Any]:
        return {"sku": self.sku, "name": self.name, "price": self.price}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Self:
        return cls(sku=data["sku"], name=data["name"], price=data["price"])


def serialize_batch(entities: list[Serializable]) -> str:
    return json.dumps([e.to_dict() for e in entities], ensure_ascii=False, indent=2)


profiles: list[Serializable] = [
    UserProfile("u1", "alice@example.com", "premium"),
    ProductListing("SKU-001", "Mechanical Keyboard", 150.0),
]
print(serialize_batch(profiles))

Bài 2: Type-safe Event System với Generic Protocol (Advanced)

Yêu cầu: Xây dựng event system sử dụng Generic Protocol:

Định nghĩa EventHandler(Protocol[E]) với method handle(self, event: E) -> None
Tạo các event types: UserCreated, OrderPlaced, PaymentProcessed
Implement EventBus class quản lý subscription và dispatch — đảm bảo handler chỉ nhận đúng event type
Viết test minh họa type safety: subscribe UserCreated handler, dispatch OrderPlaced → mypy báo lỗi

🧠 Quiz

Câu hỏi kiểm tra: @runtime_checkable Protocol kiểm tra những gì tại runtime?

[ ] A. Method names, parameter types, và return types
[x] B. Chỉ kiểm tra sự tồn tại của method names và attributes
[ ] C. Toàn bộ signature bao gồm default values
[ ] D. Method names và số lượng parameters Giải thích: @runtime_checkable sử dụng hasattr() bên dưới — chỉ kiểm tra tên method/attribute có tồn tại hay không. Nó KHÔNG kiểm tra parameter types, return types, hay số lượng parameters. Đây là lý do phải luôn kết hợp với static type checker (mypy/pyright) để có kiểm tra đầy đủ.

Xem lời giải Bài 2

python

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Protocol, TypeVar

E_contra = TypeVar("E_contra", contravariant=True)


class EventHandler(Protocol[E_contra]):
    """Generic Protocol — handler cho một loại event cụ thể."""
    def handle(self, event: E_contra) -> None: ...


@dataclass(frozen=True)
class UserCreated:
    user_id: str
    email: str
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass(frozen=True)
class OrderPlaced:
    order_id: str
    user_id: str
    total: float


class WelcomeEmailSender:
    def handle(self, event: UserCreated) -> None:
        print(f"Sending welcome email to {event.email}")

class InventoryUpdater:
    def handle(self, event: OrderPlaced) -> None:
        print(f"Updating inventory for order {event.order_id}")


E = TypeVar("E")

class EventBus:
    def __init__(self) -> None:
        self._handlers: dict[type, list[Any]] = {}

    def subscribe(self, event_type: type[E], handler: EventHandler[E]) -> None:
        self._handlers.setdefault(event_type, []).append(handler)

    def publish(self, event: E) -> None:
        for handler in self._handlers.get(type(event), []):
            handler.handle(event)


bus = EventBus()
bus.subscribe(UserCreated, WelcomeEmailSender())
bus.subscribe(OrderPlaced, InventoryUpdater())
bus.publish(UserCreated(user_id="u1", email="alice@example.com"))
bus.publish(OrderPlaced(order_id="ord-1", user_id="u1", total=99.99))

# ❌ mypy báo lỗi nếu subscribe sai type:
# bus.subscribe(UserCreated, InventoryUpdater())
# error: "InventoryUpdater" not assignable to "EventHandler[UserCreated]"

Liên kết học tiếp

Glossary keywords: structural subtyping · nominal typing · typing.Protocol · runtime_checkable · duck typing · PEP 544 · ABC vs Protocol · Generic Protocol · Interface Segregation · plugin system · covariant · contravariant

Protocols — Duck Typing có kiểm tra kiểu tĩnh Nâng cao ​

Bức tranh tư duy ​

Ổ cắm điện — Protocol trong đời thực ​

Cốt lõi kỹ thuật ​

Structural vs Nominal Typing — hai triết lý ​

typing.Protocol — định nghĩa và sử dụng ​

@runtime_checkable và isinstance() ​

Protocol với attributes, properties và class variables ​

Generic Protocols — Protocol[T] ​

Thực chiến ​

Xây dựng Plugin System cho Data Processing Framework ​

Sai lầm điển hình ​

❌ SAI #1: Kế thừa Protocol khi không cần thiết ​

❌ SAI #2: Tin tưởng runtime_checkable kiểm tra signature ​

❌ SAI #3: Fat Protocol — vi phạm Interface Segregation ​

❌ SAI #4: Nhầm Protocol với ABC — dùng sai tool ​

Under the Hood ​

mypy giải quyết structural subtyping như thế nào ​

runtime_checkable và ABCMeta.subclasshook ​

Performance implications ​

Protocol tại runtime vs type-check time ​

Checklist ghi nhớ ​

Bài tập luyện tập ​

Bài 1: Serializable Protocol (Intermediate) ​

Bài 2: Type-safe Event System với Generic Protocol (Advanced) ​

Liên kết học tiếp ​

Protocols — Duck Typing có kiểm tra kiểu tĩnh Nâng cao

Bức tranh tư duy

Ổ cắm điện — Protocol trong đời thực

Cốt lõi kỹ thuật

Structural vs Nominal Typing — hai triết lý

typing.Protocol — định nghĩa và sử dụng

@runtime_checkable và isinstance()

Protocol với attributes, properties và class variables

Generic Protocols — Protocol[T]

Thực chiến

Xây dựng Plugin System cho Data Processing Framework

Sai lầm điển hình

❌ SAI #1: Kế thừa Protocol khi không cần thiết

❌ SAI #2: Tin tưởng runtime_checkable kiểm tra signature

❌ SAI #3: Fat Protocol — vi phạm Interface Segregation

❌ SAI #4: Nhầm Protocol với ABC — dùng sai tool

Under the Hood

mypy giải quyết structural subtyping như thế nào

runtime_checkable và ABCMeta.subclasshook

Performance implications

Protocol tại runtime vs type-check time

Checklist ghi nhớ

Bài tập luyện tập

Bài 1: Serializable Protocol (Intermediate)

Bài 2: Type-safe Event System với Generic Protocol (Advanced)

Liên kết học tiếp