让 mypy 接受泛型类型的子类型作为方法参数

Question

我正在尝试将我们在代码库中使用的模式提取到更通用、可重用的结构中。但是，我似乎无法让通用类型注释与 mypy 一起使用。

这是我得到的：

from abc import (
    ABC,
    abstractmethod
)
import asyncio
import contextlib
from typing import (
    Any,
    Iterator,
    Generic,
    TypeVar
)

_TMsg = TypeVar('_TMsg')

class MsgQueueExposer(ABC, Generic[_TMsg]):

    @abstractmethod
    def subscribe(self, subscriber: 'MsgQueueSubscriber[_TMsg]') -> None:
        raise NotImplementedError("Must be implemented by subclasses")

    @abstractmethod
    def unsubscribe(self, subscriber: 'MsgQueueSubscriber[_TMsg]') -> None:
        raise NotImplementedError("Must be implemented by subclasses")


class MsgQueueSubscriber(Generic[_TMsg]):

    @contextlib.contextmanager
    def subscribe(
            self,
            msg_queue_exposer: MsgQueueExposer[_TMsg]) -> Iterator[None]:
        msg_queue_exposer.subscribe(self)
        try:
            yield
        finally:
            msg_queue_exposer.unsubscribe(self)


class DemoMsgQueSubscriber(MsgQueueSubscriber[int]):
    pass

class DemoMsgQueueExposer(MsgQueueExposer[int]):

    # The following works for mypy:

    # def subscribe(self, subscriber: MsgQueueSubscriber[int]) -> None:
    #     pass

    # def unsubscribe(self, subscriber: MsgQueueSubscriber[int]) -> None:
    #     pass

    # This doesn't work but I want it to work :)

    def subscribe(self, subscriber: DemoMsgQueSubscriber) -> None:
        pass

    def unsubscribe(self, subscriber: DemoMsgQueSubscriber) -> None:
        pass

我注释掉了一些有效但不能完全满足我的需求的代码。基本上我希望 DemoMsgQueueExposer 在其 subscribe 和 unsubscribe 方法中接受 DemoMsgQueSubscriber。如果我使用 MsgQueueSubscriber[int] 作为类型，代码类型检查得很好，但我希望它接受该类型的子类型。

我一直运行进入下面的错误

generic_msg_queue.py:55: error: Argument 1 of "subscribe" incompatible with supertype "MsgQueueExposer"

我觉得这和co-/contravariants有关系，但我尝试了好几样东西才放弃，来到这里。

Answer 1

您最好的选择是 1) 完全从 MsgQueueExposer 中删除 subscribe 和 unsubscribe，或者 2) 使 MsgQueueExposer 通用化 订阅者，作为 msg.

的补充或替代

这是方法 2 的示例，假设我们要保留 _TMsg 类型参数。请注意，为了演示目的，我添加了一个 messages() 方法：

from abc import ABC, abstractmethod
import asyncio
import contextlib
from typing import Any, Iterator, Generic, TypeVar, List

_TMsg = TypeVar('_TMsg')
_TSubscriber = TypeVar('_TSubscriber', bound='MsgQueueSubscriber')

class MsgQueueExposer(ABC, Generic[_TSubscriber, _TMsg]):

    @abstractmethod
    def subscribe(self, subscriber: _TSubscriber) -> None:
        raise NotImplementedError("Must be implemented by subclasses")

    @abstractmethod
    def unsubscribe(self, subscriber: _TSubscriber) -> None:
        raise NotImplementedError("Must be implemented by subclasses")

    @abstractmethod
    def messages(self) -> List[_TMsg]:
        raise NotImplementedError("Must be implemented by subclasses")


class MsgQueueSubscriber(Generic[_TMsg]):
    # Note that we are annotating the 'self' parameter here, so we can
    # capture the subclass's exact type.

    @contextlib.contextmanager
    def subscribe(
            self: _TSubscriber,
            msg_queue_exposer: MsgQueueExposer[_TSubscriber, _TMsg]) -> Iterator[None]:
        msg_queue_exposer.subscribe(self)
        try:
            yield
        finally:
            msg_queue_exposer.unsubscribe(self)


class DemoMsgQueSubscriber(MsgQueueSubscriber[int]):
    pass

class DemoMsgQueueExposer(MsgQueueExposer[DemoMsgQueSubscriber, int]):
    def subscribe(self, subscriber: DemoMsgQueSubscriber) -> None:
        pass

    def unsubscribe(self, subscriber: DemoMsgQueSubscriber) -> None:
        pass

    def messages(self) -> List[int]:
        pass

更广泛地说，我们想表达每个 MsgQueueExposer 仅适用于特定类型的订阅者的想法，因此我们需要在某处对该信息进行编码。

其中的一个漏洞是当您使用 MsgQueueExposer 时，mypy 将无法确保订阅者接收到的任何类型与暴露者期望的任何类型都一致。因此，如果我们将演示订阅者定义为 class DemoMsgQueSubscriber(MsgQueueSubscriber[str]) 但保持 DemoMsgQueueExposer 相同，mypy 将无法检测到此错误。

但我假设您总是要成对地创建一个新的订阅者和一个新的暴露者，并且您可以仔细审核，因此这种错误在实践中不太可能发生。

让 mypy 接受泛型类型的子类型作为方法参数

Get mypy to accept subtype of generic type as a method argument

python

generics

mypy