具有类型约束的通用 Python 映射

Question

我正在尝试用 Python 建模一个简单的消息处理系统。这个想法是根据消息类型注册充当消息处理程序的函数。像这样的东西：

class Message(Protocol):
    ...

class M1(Message):
    ...

def handler1(message: M1) -> None:
    return None

handler_container: dict[type[Message], Callable[..., None]] = {M1: handler1}

def handle_message(message: Message) -> None:
     handler = handler_container[type(message)]
     return handler(message)

这就是它的要点。当我想为每个处理程序引入不同的返回类型并通过消息类型对其进行约束时，我遇到了输入问题。想象一下：

def handle_message(message: Message[T]) -> T:
    ...

我想要的是确保当我调用

handle_message

时我得到指定的返回类型。像这样：

class M2(Message[int]):
    ...

def m2_handler(message: M2) -> int:
    ...

# or

class M3(Message[None]):
    ...

def m3_handler(message: M3) -> None:
    ...

我到了这样的地步，我注册了所有这些：

handler_container: dict[type[Message[Any]], Callable[..., Any]

然后我不得不

typing.Cast

从

handle_message

返回但是还有其他问题，比如当我

inspect.signature

处理程序时，消息返回的类型没有通过

issubclass

检查，我得到

issubclass() arg 1 must be a class

（参数是

Message[NoneType]

）。

有什么想法可以实现这种类型的安全吗？我知道可能无法完全静态地键入它，所以如果有必要，我可以接受转换。欢迎任何想法，谢谢！

Answer 1

由于您打算使用此消息处理程序容器来动态（即在运行时）添加和检索处理程序函数，您可以立即忘记完美的类型安全。

类型检查器永远无法告诉您，例如，specific 消息类型的处理程序是否已经注册。即使您对

handle_message

的调用通过了

mypy --strict

，您也永远无法从中推断出该调用是否成功。 （这只是带有额外抽象层的字典键问题。）

此外，由于类型变量需要绑定到 something，因此它们不能在 some 绑定范围之外的注释中有意义地使用。换句话说，注册表的

dict[type[Message[Any]], Callable[..., Any]

与该处理程序字典的效果差不多。

话虽这么说，如果我们已经有更高级的类型变量，将完全有可能为您的处理程序容器/注册表编写一个类，至少为您提供一个类型安全的接口在某种意义上它会

1. 强制新注册是连贯的，即注册的新处理程序将要求其参数类型和返回类型与所选消息类型相对应，并且
2. 确保 检索到的处理程序对应于其参数/返回类型中选择的消息类型（如果存在的话）。

我们可以（是的，仍然假设）实现可变映射协议的一部分（特别是

__getitem__

/

__setitem__

）来模仿您已经想到的接口并将（不太有用的注释）字典存储为受保护的属性在下面。

以下是未来可能的示例：（以下是假设性的，目前不是有效/有意义的 Python 类型。）

from collections.abc import Callable
from typing import Any, Protocol, TypeVar

T = TypeVar("T")
M = TypeVar("M", bound="Message[T]", args=1)


class Message(Protocol[T]):
    x: T


class HandlerContainer:
    _registry: dict[type[Message[Any]], Callable[[Any], Any]] = {}

    def __setitem__(
        self,
        msg_type: type[M[T]],
        handler_function: Callable[[M[T]], T]
    ) -> None:
        self._registry[msg_type] = handler_function

    def __getitem__(
        self,
        msg_type: type[M[T]],
    ) -> Callable[[M[T]], T]:
        return self._registry[msg_type]


handler_container = HandlerContainer()


def handle_message(message: Message[T]) -> T:
    handler = handler_container[(type(message))]
    return handler(message)

---

但不幸的是我们没有更高种类的类型变量，所以现在，恐怕你不能做更多的事情：

from collections.abc import Callable
from typing import Any, Protocol, TypeVar, cast

T = TypeVar("T")


class Message(Protocol[T]):
    x: T


handler_container: dict[type[Message[Any]], Callable[[Any], Any]] = {}


def handle_message(message: Message[T]) -> T:
    handler = handler_container[(type(message))]
    return cast(T, handler(message))

用法：

from typing import Generic, TypeVar
# ... import Message, handler_container

U = TypeVar("U")


class M1(Generic[U]):
    x: U

def handler1(message: M1[U]) -> U:
    raise NotImplementedError

class M2:
    x: int

def handler2(message: M2) -> int:
    raise NotImplementedError

class M3:
    x: None

def handler3(message: M3) -> None:
    raise NotImplementedError

handler_container[M1] = handler1
handler_container[M2] = handler2
# handler_container[M3] = handler3

output1 = handle_message(M1[str]())
output2 = handle_message(M2())
output3 = handle_message(M3())
reveal_locals()

Mypy 输出：

note: Revealed local types are:
note:     output1: builtins.str
note:     output2: builtins.int
note:     output3: <partial None>

（注意 Mypy 无法检测到我们从未添加过

M3

处理程序。）