使用策略模式时沉默 mypy arg-type 错误

Question

最小示例：

from typing import overload, TypeVar, Generic


class EventV1:
    pass


class EventV2:
    pass


class DataGathererV1:
    def process(self, event: EventV1):
        pass

    def process2(self, event: EventV1):
        pass


class DataGathererV2:
    def process(self, event: EventV2):
        pass

    def process2(self, event: EventV2):
        pass


class Dispatcher:
    def __init__(self):
        self.worker_v1: DataGathererV1 = DataGathererV1()
        self.worker_v2: DataGathererV2 = DataGathererV2()

    def dispatch(self, event: EventV1 | EventV2):
        handler: DataGathererV1 | DataGathererV2 = self.worker_v1 if isinstance(event, EventV1) else self.worker_v2

        handler.process(event)

        # Common logic

        handler.process2(event)

        # Common logic

        handler.process(event)

        # etc...

在上面的代码中，我使用一种“策略”模式来处理事件。我想避免将我的

dispatch

方法拆分为两种方法，因为我认为这没有意义并且会生成代码重复。

Mypy 给了我以下错误，我不知道如何正确键入我的代码以避免此类错误。

example.py:36: error: Argument 1 to "process" of "DataGathererV1" has incompatible type "EventV1 | EventV2"; expected "EventV1"  [arg-type]
example.py:36: error: Argument 1 to "process" of "DataGathererV2" has incompatible type "EventV1 | EventV2"; expected "EventV2"  [arg-type]
example.py:40: error: Argument 1 to "process2" of "DataGathererV1" has incompatible type "EventV1 | EventV2"; expected "EventV1"  [arg-type]
example.py:40: error: Argument 1 to "process2" of "DataGathererV2" has incompatible type "EventV1 | EventV2"; expected "EventV2"  [arg-type]
example.py:44: error: Argument 1 to "process" of "DataGathererV1" has incompatible type "EventV1 | EventV2"; expected "EventV1"  [arg-type]
example.py:44: error: Argument 1 to "process" of "DataGathererV2" has incompatible type "EventV1 | EventV2"; expected "EventV2"  [arg-type]

我尝试添加一个重载函数来同时检索我的处理程序和事件，以强制类型“链接”（见下文），但是它并没有解决我的问题。

    @overload
    def __handler(self, event: EventV1) -> tuple[DataGathererV1, EventV1]:
        ...

    @overload
    def __handler(self, event: EventV2) -> tuple[DataGathererV2, EventV2]:
        ...

    def __handler(self, event: EventV1 | EventV2) -> tuple[DataGathererV1 | DataGathererV2, EventV1 | EventV2]:
        return (
            self.worker_v1 if isinstance(event, EventV1) else self.worker_v2,
            event
        )

我的“梦想”解决方案向 mypy 表明

event

的类型和

handler

的类型“链接”在一起。

我想避免的是将每个

DataGathererVX

中事件的输入类型设置为

EventV1 | EventV2

并在每个方法的开头添加

assert isinstance(event, EventVX)

，因为我的目标是在调用这些方法时出现错误事件类型不正确。

Answer 1

您需要构建代码以利用类型缩小，以便可以在

self.worker_v1

的

特定

类型可以从self.worker_v2缩小到

event

的上下文中执行

EventV1 | EventV2

或

EventV1

或

EventV2

。除了

handler

本身不知道是

DataGatherV1

或

DataGatherV2

之外，任一类的

process

方法只能处理其特定类型，而不是

EventV1 | EventV2

类型的值。

def dispatch(self, event: EventV1 | EventV2):
    if isinstance(event, EventV1):
        # static type of event is inferred to be EventV1
        # if we reach this code.

        self.worker_v1.process(event)
        self.worker_v1.process2(event)
        self.worker_v1.process(event)


    else:
        # static type of event is inferred to be EventV2
        # if we reach this code.
        self.worker_v2.process(event)
        self.worker_v2.process2(event)
        self.worker_v2.process(event)

您想要进行的因式分解在静态类型下根本不起作用。不过，您可能还想看看

functools.singledispatch

。