如何创建包含 threading.Event 和 multiprocessing.Event 的协议？

Question

在 Python 标准库中 multiprocessing.Event 被明确声明为 threading.Event 的克隆并具有相同的接口。我想注释变量和参数，以便它们可以接受这些 class 中的任何一个，并且 mypy 会对它们进行类型检查。我尝试创建一个协议（我使用 multiprocessing.synchronize.Event 因为那是 multiprocessing.Event 返回的实际 class）。

import multiprocessing
import threading

from typing import Optional, Type, Protocol


class Event(Protocol):
    def wait(self, timeout: Optional[float]) -> bool:
        ...

    def set(self) -> None:
        ...

    def clear(self) -> None:
        ...

    def is_set(self) -> bool:
        ...


class Base:
    flag_class: Type[Event]

    def foo(self, e: Event):
        pass


class DerivedOne(Base):
    flag_class = multiprocessing.synchronize.Event

    def foo(self, e: multiprocessing.synchronize.Event):
        pass


class DerivedTwo(Base):
    flag_class = threading.Event

    def foo(self, e: threading.Event):
        pass

但是，mypy（版本 0.761）无法识别 multiprocessing.Event 和 threading.Event 都实现了我定义的协议：

$ mypy proto.py

proto.py:31: error: Argument 1 of "foo" is incompatible with supertype "Base"; supertype defines the argument type as "Event"
proto.py:38: error: Argument 1 of "foo" is incompatible with supertype "Base"; supertype defines the argument type as "Event"
Found 2 errors in 1 file (checked 1 source file)

为什么 mypy 无法识别我的协议，我该如何解决？

Answer 1

这不是 Protocol 问题。您将 foo 的签名更改为 更严格的变体 。 Base.foo() 接受 any Event 实现，而你的每个子classes 只接受一个具体的实现。这违反了 Liskov substitution principle 并且 Mypy 不允许这样做是正确的。

您必须在此处使用绑定 TypeVar 和 Generic 的组合，因此您可以创建 Base 的不同具体子 class不同类型：

from typing import Generic, Optional, Protocol, Type, TypeVar

# Only things that implement Event will do
T = TypeVar("T", bound=Event)

# Base is Generic, subclasses need to state what exact class
# they use for T; as long as it's an Event implementation, that is.
class Base(Generic[T]):
    flag_class: Type[T]

    def foo(self, e: T) -> None:
        pass

这实际上使 Base 成为一种模板 class，有了 T 模板插槽，您可以插入任何东西，只要 任何东西 实现你的协议。这也更加稳健，因为您现在不会意外混淆 Event 实现（将 threading.Event 和 multiprocessing.Event 组合在一个子 class 中）。

因此，以下两个不同 Event 实现的子 class 是正确的：

class DerivedOne(Base[multiprocessing.synchronize.Event]):
    flag_class = multiprocessing.synchronize.Event

    def foo(self, e: multiprocessing.synchronize.Event) -> None:
        pass

class DerivedTwo(Base[threading.Event]):
    flag_class = threading.Event

    def foo(self, e: threading.Event) -> None:
        pass

但使用未实现协议方法的 class 是错误的：

# Mypy flags the following class definition as an error, because a lock
# does not implement the methods of an event.
# error: Type argument "threading.Lock" of "Base" must be a subtype of "proto.Event"
class Wrong(Base[threading.Lock]):
    flag_class = threading.Lock

    def foo(self, e: threading.Lock) -> None:
        pass

混合类型也是错误的：

# Mypy flags 'def foo' as an error because the type it accepts differs from
# the declared type of the Base[...] subclass
# error: Argument 1 of "foo" is incompatible with supertype "Base"; supertype defines the argument type as "Event"
class AlsoWrong(Base[threading.Event]):
    flag_class = threading.Event

    def foo(self, e: multiprocessing.synchronize.Event) -> None:
        pass

# Mypy flags 'flag_class' as an error because the type differs from the
# declared type of the Base[...] subclass
# error: Incompatible types in assignment (expression has type "Type[multiprocessing.synchronize.Event]", base class "Base" defined the type as "Type[threading.Event]")
class StillWrong(Base[threading.Event]):
    flag_class = multiprocessing.synchronize.Event

    def foo(self, e: threading.Event) -> None:
        pass

如何创建包含 threading.Event 和 multiprocessing.Event 的协议？

How can I create a Protocol that encompases both threading.Event and multiprocessing.Event?

python

type-hinting

mypy