Получите 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"

Я чувствую, что это как-то связано с соавторами / контравариантами, но я попробовал несколько вещей, прежде чем сдался и пришел сюда.

Answer 1

Ваши лучшие ставки: 1) просто удалите subscribe и unsubscribe из MsgQueueExposer в целом, или 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 работает только для определенного типа абонента, поэтому нам нужно было где-то кодировать эту информацию.

Единственная дыра в этом заключается в том, что mypy не сможет убедиться, что при использовании MsgQueueExposer, что любой тип, который получает подписчик, и что любой тип, ожидаемый экспозитором, согласится. Таким образом, если мы определим демо-подписчика как class DemoMsgQueSubscriber(MsgQueueSubscriber[str]), но оставим DemoMsgQueueExposer таким же, mypy не сможет обнаружить эту ошибку.

Но я предполагаю, что вы всегда будете создавать нового подписчика и нового разоблачителя в парах, и это то, что вы можете тщательно проверить, поэтому на практике эта ошибка вряд ли произойдет.