Как я могу создать протокол, который включает в себя как threading.Event, так и multiprocessing.Event? - PullRequest
Новая
       41

Как я могу создать протокол, который включает в себя как threading.Event, так и multiprocessing.Event?

2 голосов
/ 16 января 2020

В стандартной библиотеке Python явно указано, что multiprocessing.Event является клоном threading.Event и имеет тот же интерфейс. Я хотел бы аннотировать переменные и аргументы, чтобы они могли принимать любой из этих классов, и mypy проверил бы их тип. Я попытался создать протокол (я использовал multiprocessing.synchronize.Event, поскольку это фактический класс, возвращаемый multiprocessing.Event). 

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 не распознает мой протокол и как я могу его исправить?

1 Ответ

2 голосов
/ 16 января 2020

Это не проблема Protocol. Вы изменяете подпись от foo до более строгих вариантов . Base.foo() принимает любую Event реализацию, в то время как каждый из ваших подклассов принимает только одну конкретную реализацию. Это нарушение принципа подстановки Лискова , и Mypy правильно, если не допустит этого.

Здесь вам придется использовать комбинацию связанных TypeVar и Generic, поэтому Вы можете создать различные конкретные подклассы Base, которые принимают другой тип: 

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 своего рода классом шаблона, с T слотом шаблона, в который вы можете подключить что угодно, до тех пор, пока что-нибудь реализует ваш протокол. Это также намного надежнее, поскольку теперь вы не можете случайно смешать реализации Event (объединяя threading.Event и multiprocessing.Event в одном подклассе).

Таким образом, следующие подклассы для двух разных * Реализации 1028 * верны: 

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

, но использование класса, который не реализует методы протокола, является ошибкой: 

# 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
...