Python typing.Protocol：写库时拥抱 duck typing 又有类型提示

起因

写一个数据导出库，接收任何"长得像 file-like 对象的"输入：
内置 open() 返回值、io.BytesIO、Django 的 UploadedFile、
S3 client 返回的 streaming body……

def export(stream):
    stream.write(b'header')
    for row in data:
        stream.write(row.serialize())

类型怎么标？stream: BinaryIO？只 cover 标准库；stream: object？
失去类型提示。

typing.Protocol（PEP 544）解决：定义"结构子类型"（structural subtyping），
描述"具有 X 方法的任何对象"，无需对方继承。

解决方案

from typing import Protocol

class WritableBytes(Protocol):
    def write(self, data: bytes) -> int: ...
    def flush(self) -> None: ...

def export(stream: WritableBytes) -> None:
    stream.write(b'header')
    for row in data:
        stream.write(row.serialize())
    stream.flush()

任何"有 write(bytes) → int 和 flush() → None" 方法的类都满足。
不需要 import 我的 Protocol、不需要 inherit、不需要 register。

类型检查器（mypy / pyright）静态确认：

import io
export(io.BytesIO())               # ✅
export(open('out.bin', 'wb'))      # ✅
export("not a file")                # ❌ mypy 报错

runtime check

需要在运行时判断："这东西满足 Protocol 吗？"

from typing import Protocol, runtime_checkable

@runtime_checkable
class WritableBytes(Protocol):
    def write(self, data: bytes) -> int: ...

if isinstance(obj, WritableBytes):
    obj.write(b'x')

@runtime_checkable 让 isinstance 工作。但只检查方法存在，
不检查方法签名 —— 比静态检查弱。

实战例子：可插拔的 storage backend

from typing import Protocol

class Storage(Protocol):
    def get(self, key: str) -> bytes | None: ...
    def set(self, key: str, value: bytes, ttl: int | None = None) -> None: ...
    def delete(self, key: str) -> None: ...

class InMemoryStorage:
    def __init__(self):
        self._d: dict[str, bytes] = {}
    def get(self, key): return self._d.get(key)
    def set(self, key, value, ttl=None): self._d[key] = value
    def delete(self, key): self._d.pop(key, None)

class RedisStorage:
    def __init__(self, url): self._r = redis.from_url(url)
    def get(self, key): return self._r.get(key)
    def set(self, key, value, ttl=None):
        if ttl: self._r.set(key, value, ex=ttl)
        else: self._r.set(key, value)
    def delete(self, key): self._r.delete(key)

def setup_cache(storage: Storage) -> Cache:
    return Cache(storage)

setup_cache(InMemoryStorage())
setup_cache(RedisStorage('redis://localhost'))

InMemoryStorage / RedisStorage 都没 inherit Storage，
但都 conform 该结构 → mypy 通过。

与 ABC 对比

from abc import ABC, abstractmethod

class Storage(ABC):
    @abstractmethod
    def get(self, key: str) -> bytes | None: ...
    @abstractmethod
    def set(self, key: str, value: bytes, ttl: int | None = None) -> None: ...
    @abstractmethod
    def delete(self, key: str) -> None: ...

class InMemoryStorage(Storage):    # 必须显式继承
    ...

ABC 强制继承。Protocol 不强制 → 第三方库的类不需要修改就能用。

适用场景：

Protocol：写库 / interface 定义，鸭子类型友好
ABC：内部类型层次、要 share 实现（Mixin）、强制 inherit

generic protocol

from typing import Protocol, TypeVar

T = TypeVar('T', covariant=True)

class Iterable(Protocol[T]):
    def __iter__(self) -> 'Iterator[T]': ...

class Iterator(Protocol[T]):
    def __next__(self) -> T: ...

实际 Iterable 已在 typing 模块，举例说明语法。

在标准库里你已经在用

typing / collections.abc 模块里很多就是 Protocol：

from collections.abc import Iterable, Mapping, Hashable, Container
from typing import Protocol, SupportsLen, SupportsInt

SupportsLen 是 def __len__(self) -> int。所以 len(x) 能用的
都满足。

给现有类"贴" Protocol

from third_party import SomeClass

class HasFoo(Protocol):
    def foo(self) -> str: ...

# SomeClass 有 foo() 方法但作者没标注
x: HasFoo = SomeClass()    # mypy 检查 OK
x.foo()

零侵入。

实战 case：测试替身

class Notifier(Protocol):
    def send(self, msg: str, to: str) -> None: ...

def process_order(order: Order, notifier: Notifier) -> None:
    # ...
    notifier.send(f'Order {order.id} confirmed', to=order.email)

# 生产
process_order(order, EmailNotifier())

# 测试
class FakeNotifier:
    def __init__(self):
        self.sent: list[tuple[str, str]] = []
    def send(self, msg, to):
        self.sent.append((msg, to))

def test_confirm():
    n = FakeNotifier()
    process_order(test_order, n)
    assert ('Order 1 confirmed', '[email protected]') in n.sent

FakeNotifier 不需要 inherit Notifier ABC，纯写实现即可。

与 `__class_getitem__` / `TypeVar` 联用

from typing import Protocol, TypeVar

K = TypeVar('K')
V = TypeVar('V')

class Cache(Protocol[K, V]):
    def get(self, key: K) -> V | None: ...
    def set(self, key: K, value: V) -> None: ...

class StringIntCache:
    def get(self, key: str) -> int | None: ...
    def set(self, key: str, value: int) -> None: ...

def use(c: Cache[str, int]):
    v = c.get('x')   # mypy 知道 v: int | None

效果

库的 API 类型严格但不要求用户继承
测试时随便造 fake，不需要 mock 框架
重构内部实现时 Protocol 是"接口"，业务代码改少
mypy / pyright 在 IDE 里实时提示，写错立刻知道

踩过的坑

Protocol 不能 instantiate：x = WritableBytes() 报错（没意义，
它是接口）。
runtime_checkable 检查只看方法名：isinstance(obj, WritableBytes)
只确认有 write 和 flush，不查签名。运行时碰到方法签名不对仍
crash。
Protocol 的方法有 default impl 让它变成 mixin 又不强制继承：
```python
class Repr(Protocol):
def repr(self) -> str: ...

class Mixin(Repr):
def repr(self):
return f'<{type(self).name}>'
```
语义微妙，团队约定清楚。

structural subtyping 太宽容：所有有 read() 方法的都被当
readable，比如自定义类 class Sensor: def read(self): ... 不该
是 file-like 但 mypy 不报错。给 Protocol 多放几个方法（read +
readable 等）让匹配更严。
Protocol 间不能继承 default impl：跟 mixin 不同。要复用代码用
ABC + Protocol 组合，或者纯函数化。