获取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接受DemoMsgQueSubscriber它subscribe和unsubscribe方法.如果我MsgQueueSubscriber[int]用作类型,代码类型检查就好了,但我希望它接受它的子类型.

我一直遇到以下错误.

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

我觉得这与co/contravariants有关,但在我放弃之前我尝试了几件事来到这里.

Answer 1

您最好的选择是 1) 完全删除和subscribe，或者 2)对订阅者进行通用化，除了.unsubscribeMsgQueueExposerMsgQueueExposermsg

下面是方法 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内容仅适用于特定类型的订阅者，因此我们需要在某处对该信息进行编码。

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

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