如何为消息总线subscribers添加正确TypeVar类型提示及相关优化？

我开发了一个基于消息总线传递消息的简单应用，为实现严格类型检查，我以BaseMessage子类的实例作为消息载体，并要求处理器仅接收特定子类作为参数。目前代码已能通过pyright检测出错误订阅（例如将处理MessageType2的处理器绑定到MessageType1主题），但无法为MessageBus的subscribers字典添加正确的类型提示：

• 尝试self.subscribers: dict[Type[Message], list[Callable[[Message], None]]] = {}时，pyright报错Type Variable "Message" has no meaning in this context；
• 尝试self.subscribers: dict[Type[BaseMessage], list[Callable[[BaseMessage], None]]] = {}时，append操作报错Type "(Message@subscribe) -> None" is not assignable to type "(BaseMessage) -> None"。

核心问题 #

1. 上述代码中self.subscribers的正确类型提示是什么？为什么？
2. 是否可以定义类型提示别名（如MessageHandler = Callable[[Message], None]）提升代码可读性？
3. 有没有基于Python标准库的、传递消息对象的发布订阅模型推荐方案？

现有工作代码 #

from dataclasses import dataclass, field, replace
from typing import Type, TypeVar, Callable

class BaseMessage:
    # 抽象基类，所有消息必须继承自此类
    pass

# Message代表BaseMessage的任意子类，用于类型提示
Message = TypeVar('Message', bound=BaseMessage)

# 定义具体消息类型
@dataclass
class MessageType1(BaseMessage):
    data: int

@dataclass
class MessageType2(BaseMessage):
    point: tuple[float, float]


class MessageBus():
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, messagetype: Type[Message], handler: Callable[[Message], None]) -> None:
        """为指定消息类型注册处理器（订阅者）"""
        self.subscribers.setdefault(messagetype, []).append(handler)

    def publish(self, message: BaseMessage):
        """将消息发送给对应类型的所有处理器"""
        for handler in self.subscribers.get(type(message), []):
            handler(message)


# 使用示例
def my_handler1(message: MessageType1):
    print(f"my_handler1 message: {message}")

def my_handler2(message: MessageType2):
    print(f"my_handler2 message: {message}")

bus = MessageBus()
bus.subscribe(MessageType1, my_handler1)
bus.subscribe(MessageType2, my_handler2)
bus.subscribe(MessageType1, my_handler2)  # 期望此处被检测出类型错误
bus.publish(MessageType1(3))
bus.publish(MessageType2((5.0, 2.5)))

pyright检测到的错误信息 #

error: Argument of type "(message: MessageType2) -> None" cannot be assigned to parameter "handler" of type "(Message@subscribe) -> None" in function "subscribe"
   Type "(message: MessageType2) -> None" is not assignable to type "(MessageType1) -> None"
    Parameter 1: type "MessageType1" is incompatible with type "MessageType2"
      "MessageType1" is not assignable to "MessageType2" (reportArgumentType)

问题解答 #

1. self.subscribers的正确类型提示及原因

正确的类型提示需要结合**逆变TypeVar**解决Callable的类型兼容性问题，代码如下：

from typing import TypeVar, Callable, Type, dict

# 定义逆变的TypeVar，允许子类处理器适配父类类型
Message = TypeVar('Message', bound=BaseMessage, contravariant=True)

class MessageBus():
    def __init__(self):
        self.subscribers: dict[Type[BaseMessage], list[Callable[[BaseMessage], None]]] = {}

    def subscribe(self, messagetype: Type[Message], handler: Callable[[Message], None]) -> None:
        self.subscribers.setdefault(messagetype, []).append(handler)

原因：
Callable的参数类型是逆变的——如果A是B的子类，那么Callable[[B], None]是Callable[[A], None]的子类。通过给Message添加contravariant=True标记，类型检查器会认可Callable[[MessageSubclass], None]可以赋值给Callable[[BaseMessage], None]，同时保留对错误订阅的检测能力。

如果不想修改TypeVar的逆变属性，也可以用cast辅助类型检查器理解业务逻辑（因为我们的publish方法只会给处理器传递对应类型的消息，强制转换是安全的）：

from typing import cast

class MessageBus():
    def __init__(self):
        self.subscribers: dict[Type[BaseMessage], list[Callable[[BaseMessage], None]]] = {}

    def subscribe(self, messagetype: Type[Message], handler: Callable[[Message], None]) -> None:
        self.subscribers.setdefault(messagetype, []).append(cast(Callable[[BaseMessage], None], handler))

2. 可以定义类型提示别名提升可读性

完全可以，而且非常推荐。你可以结合逆变TypeVar定义通用的处理器别名，让代码更简洁易读：

Message = TypeVar('Message', bound=BaseMessage, contravariant=True)
MessageHandler = Callable[[Message], None]

class MessageBus():
    def __init__(self):
        self.subscribers: dict[Type[BaseMessage], list[MessageHandler[BaseMessage]]] = {}

    def subscribe(self, messagetype: Type[Message], handler: MessageHandler[Message]) -> None:
        self.subscribers.setdefault(messagetype, []).append(handler)

后续如果需要调整处理器的类型定义（比如添加返回值参数），只需要修改MessageHandler的别名即可，无需修改多处代码。

3. 基于Python标准库的发布订阅方案

Python标准库没有开箱即用的发布订阅框架，但可以基于以下模块实现符合需求的模型：

• queue.Queue：为每个消息类型创建独立队列，订阅者监听对应队列，发布者将消息推入目标队列。适合单线程或多线程同步场景，实现简单直观。
• asyncio.Queue：针对异步IO场景设计，用法与queue.Queue类似，但支持异步等待消息，适配异步应用架构。
• multiprocessing.Queue/Pipe：如果需要跨进程通信，可以用这些组件实现分布式的发布订阅模型，适合多进程场景。
• collections.defaultdict：结合列表实现类似你当前的内存型发布订阅模型，无需额外依赖，仅用标准库即可完成，适合轻量级场景。

需要注意的是，标准库实现的模型默认没有类型检查能力，你需要自行添加类型注解来保证类型安全。

内容的提问来源于stack exchange，提问作者MacFreek