CODEWITHSUNDEEP
pythontype-hintingmypy
sagittarian
sagittarian

Reputation: 1094

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

In the Python standard library multiprocessing.Event is explicitly stated to be a clone of threading.Event and to have the same interface. I would like to annotate variables and arguments so that they can accept either of those classes, and mypy would type-check them. I tried creating a Protocol (I used multiprocessing.synchronize.Event since that is the actual class that is returned by 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

However, mypy (version 0.761) doesn't recognize that multiprocessing.Event and threading.Event both implement the protocol I defined:

$ 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)

Why doesn't mypy recognize my protocol and how can I fix it?

Upvotes: 3

Views: 773

Answers (1)

Martijn Pieters
Martijn Pieters

Reputation: 1123700

This is not a Protocol issue. You altering the signature of foo to stricter variants. Base.foo() accepts any Event implementation, while your subclasses each only accept a single concrete implementation. That's a violation of the Liskov substitution principle and Mypy is correct in not allowing this.

You'd have to use a combination of a bound TypeVar and Generic here, so you can create different concrete subclasses of Base that take a different type:

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

This essentially makes Base a kind of template class, with T the template slot you can plug anything into, as long as that anything implements your protocol. This is also far more robust as you now can't accidentally mix up the Event implementations (combining threading.Event and multiprocessing.Event in a single subclass).

So the following subclasses for the two different Event implementations are correct:

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

but using a class that doesn't implement the protocol methods is an error:

# 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

It is also an error to mix the types:

# 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

Upvotes: 5

Related Questions