Python Protocol 怎样描述带异步方法的结构类型

给异步仓储、HTTP 客户端或插件接口做类型约束时，常见需求是“只要对象拥有指定的异步方法，就可以传进来”，而不是要求所有实现都继承同一个基类。typing.Protocol 正适合这种结构类型：类型检查器比较对象实际提供的成员和签名，不要求显式继承。

直接答案：如果调用方写的是 user = await source.fetch(id)，协议通常应声明 async def fetch(...) -> User: ...。这里的 User 表示 await 之后 的结果，不需要再包一层 Awaitable[User]。

官方文档：https://docs.python.org/3/library/typing.html#typing.Protocol

现象：方法明明存在，类型检查仍报不兼容

假设业务层只关心“能按用户 ID 异步取回用户”的对象。一个实现从数据库读取，另一个实现调用远程服务，它们不需要共享父类。

from dataclasses import dataclass
from typing import Protocol

@dataclass(frozen=True)
class User:
    id: int
    name: str

class UserSource(Protocol):
    async def fetch(self, user_id: int) -> User:
        # 省略号表示协议只描述成员契约
        ...

async def show_name(source: UserSource, user_id: int) -> str:
    # fetch 的调用结果可等待，await 后得到 User
    user = await source.fetch(user_id)
    return user.name

任何提供兼容 fetch() 签名的类都能作为 UserSource，无需写 class SqlUserSource(UserSource)。这就是静态鸭子类型：类型检查器关心结构，不关心名义继承。

第一层检查：async def 的返回注解表示什么

async def fetch(...) -> User 并不是说调用 fetch() 立即得到 User。调用会创建协程对象，await 该对象后才得到 User。因此协议里的返回注解写业务结果类型即可。

class SqlUserSource:
    async def fetch(self, user_id: int) -> User:
        # 真实项目可以在这里等待异步数据库驱动
        await asyncio.sleep(0)
        return User(id=user_id, name="Ada")

source: UserSource = SqlUserSource()
# 这行赋值是静态契约哨兵；实现类无需继承协议

若编辑器把 source.fetch(1) 推断为 Coroutine[Any, Any, User]，这是正常现象；协程实现了可等待协议，await 后结果才是 User。

第二层检查：不要给 async def 多包一层 Awaitable

最常见的错误是把协议写成 async def fetch(...) -> Awaitable[User]。这个签名的含义不是“fetch 可等待”，而是“await fetch 后还会得到另一个 Awaitable”。类型检查器因此可能把最终值推断成还需要再 await 一次的对象。

from collections.abc import Awaitable

class WrongSource(Protocol):
    async def fetch(self, user_id: int) -> Awaitable[User]:
        # 错误契约：第一次 await 后仍要求得到 Awaitable[User]
        ...

async def wrong_use(source: WrongSource) -> User:
    pending_user = await source.fetch(1)
    # pending_user 仍是 Awaitable[User]，需要第二次 await 才是 User
    return await pending_user

只有当异步函数确实返回另一个延迟对象时才这样声明。普通异步接口直接写 async def -> User。

什么时候使用 def 返回 Awaitable

有时接口想表达的是“调用后返回任意可等待对象”，但并不要求实现必须用 async def。此时可以把协议成员写成普通 def，返回 Awaitable[T]。这样实现可以返回协程、Future 或自定义 Awaitable。

from collections.abc import Awaitable

class DeferredUserSource(Protocol):
    def fetch(self, user_id: int) -> Awaitable[User]:
        # 只约束调用结果可等待，不约束实现采用 async def
        ...

class TaskUserSource:
    def fetch(self, user_id: int) -> asyncio.Task[User]:
        # 把协程调度成 Task，Task 也是 Awaitable[User]
        return asyncio.create_task(self._load(user_id))

    async def _load(self, user_id: int) -> User:
        # 私有协程负责实际异步加载
        await asyncio.sleep(0)
        return User(user_id, "Grace")

选择标准很简单：要强调“这是异步方法”，用协议中的 async def；要允许任何返回可等待对象的调用形式，用 def -> Awaitable[T]。两者的调用端都可以 await，但接口约束范围不同。

第三层检查：参数名和参数种类也是协议的一部分

结构匹配不只比较“有几个参数”。如果协议允许关键字调用，参数名就会影响兼容性；位置专用参数、关键字专用参数和默认值也会影响调用集合。调用方能做的每一种合法调用，实现方都必须接得住。

class UserSource(Protocol):
    async def fetch(
        self,
        user_id: int,
        /,
        *,
        timeout: float | None = None,
    ) -> User:
        # user_id 仅限位置传入，timeout 仅限关键字传入
        ...

class HttpUserSource:
    async def fetch(
        self,
        user_id: int,
        /,
        *,
        timeout: float | None = None,
    ) -> User:
        # 参数种类和可接受调用方式与协议保持一致
        return await self._request(user_id, timeout=timeout)

如果实现把 timeout 改成必填位置参数，调用方按协议写 fetch(1, timeout=0.5) 时就会失败，所以类型检查器拒绝这种实现是有意义的。

第四层检查：异步泛型协议如何保留结果类型

加载器结构相同但结果类型不同，可以把协议做成泛型。Python 3.11 及更早版本可使用 TypeVar；较新的 Python 也支持类型参数语法。只读返回值通常可以使用协变类型变量。

from typing import Protocol, TypeVar

T_co = TypeVar("T_co", covariant=True)

class AsyncLoader(Protocol[T_co]):
    async def load(self, key: str) -> T_co:
        # T_co 表示 await 后的结果类型
        ...

async def load_user(loader: AsyncLoader[User]) -> User:
    # 类型检查器能够保留具体的 User 结果
    return await loader.load("current-user")

如果类型变量同时出现在输入和输出位置，不要机械地标记为协变。此时通常应使用不变类型变量，或把读写能力拆成两个更小的协议。

第五层检查：异步回调应使用 __call__ 协议

Callable[[Event], Awaitable[None]] 能描述简单异步回调；若回调包含关键字专用参数、重载或更复杂调用形状，应定义带 __call__ 的 Protocol。

from typing import Protocol

class EventHandler(Protocol):
    async def __call__(
        self,
        event: User,
        *,
        retry: bool = False,
    ) -> None:
        # 协议同时约束异步结果和关键字参数 retry
        ...

async def dispatch(handler: EventHandler, event: User) -> None:
    # 调用对象本身，await 后没有业务返回值
    await handler(event, retry=True)

函数、实现了异步 __call__ 的对象都可以匹配，只要完整签名可赋值。Protocol 不要求实现使用同一个类层次。

第六层检查：runtime_checkable 不能验证异步签名

@runtime_checkable 只适合做很浅的成员存在检查。官方文档明确说明，它不会检查成员的类型签名。一个对象只要有同名 fetch 属性，就可能通过 isinstance()，即使该方法是同步的、参数不兼容或返回错误类型。

from typing import Protocol, runtime_checkable

@runtime_checkable
class RuntimeSource(Protocol):
    async def fetch(self, user_id: int) -> User:
        # 装饰器只让协议可用于 isinstance，不验证完整签名
        ...

class MisleadingSource:
    def fetch(self, user_id: str) -> int:
        # 名称存在，但参数、返回值和异步性质都不兼容
        return 1

assert isinstance(MisleadingSource(), RuntimeSource)
# 运行时结构检查可能为真，静态类型检查仍应报错

因此，异步 Protocol 的主验证手段应是 mypy、Pyright 等静态检查。运行时若必须防御外部插件，可以在边界处调用后使用 inspect.isawaitable() 检查结果，并把不兼容异常转换为明确的插件错误；但这不能替代静态签名检查。

用静态哨兵和最小运行测试反向确认

静态检查负责证明结构兼容，运行测试负责证明实现真的可等待并返回期望值。两类测试不要混在一起。

def accepts_source(source: UserSource) -> None:
    # 该函数只用于触发静态结构检查
    pass

async def smoke_test() -> None:
    source = SqlUserSource()
    accepts_source(source)
    # 运行测试确认协程可等待且结果符合业务预期
    user = await source.fetch(7)
    assert user.id == 7

若实现来自第三方库且缺少注解，可以为它写一个类型安全的适配器，而不是在业务代码里到处使用 Any 或 cast()。适配器既统一签名，也集中处理超时、异常与返回数据转换。

最终检查清单

常见问题

实现类必须继承 Protocol 吗？

不必须。成员和签名兼容即可隐式实现。显式继承适合希望类型检查器更早检查实现完整性，或需要复用协议默认实现的场景。

Protocol 能保证方法运行时一定是协程函数吗？

静态检查器可以根据签名检查调用结果是否可等待；runtime_checkable 本身不能验证异步性质或返回类型。

Callable 和带 __call__ 的 Protocol 怎么选？

简单参数列表用 Callable[..., Awaitable[T]] 即可；需要关键字专用参数、重载或精确参数名时，用 __call__ Protocol。

为什么 async def 返回 T，而不是 Coroutine[T]？

返回注解描述协程执行完成后的结果。调用表达式的类型由类型检查器展开为协程类型，业务代码 await 后得到 T。

描述带异步方法的结构类型，最稳妥的起点是把调用方写出来：如果调用方需要 await obj.method() 后得到 T，就在 Protocol 中声明 async def method() -> T。出现不兼容时，再按 Awaitable 层级、参数形状、泛型方向和运行时边界逐层排查。