是否有支持对象消息传递的 Python 包？ [关闭]

Question

我有兴趣使我当前的 Python 项目符合 Alan Kay 最初构想的面向对象范式——from my understanding 是关于对象之间的消息传递，就像它是关于对象一样自己。

我正在考虑的那种包中的示例代码可能如下所示：

def do_laundry(self):
    send(self.car, "DRIVE_TO", (self.laundromat))
    washing_machine = self.look_for(WashingMachine)
    send(self.washing_machine, "LOAD")
    ...

我可以看到紧密地坚持这种范式可以如何简化对象接口，而且消息传递的基础知识似乎并不难实现，所以我很惊讶我找不到任何相关的东西。 （我发现的所有与消息相关的包都与电子邮件或其他 Internet 活动有关。）是否有一个包可以满足我的需求，或者我误认为它对 Python 有用？可以想象，尚未完成并且可能有用，但我过去的经验表明这不太可能。

附：我知道this question，但它有不同的重点，它已经有 10 年没有活跃了，它所引用的 Python 特定资源的链接已损坏。

Answer 1

您可以为此类工作使用 pub 子系统。我是rxpy 之类的工具的忠实粉丝，但 Pub-Sub 的核心概念很容易添加自己。以此为例：

class DriveTo:
    def drive_to(self, location):
        print(f"Driving To {location}")


class LoadWashingMachine:
    def load_washing_machine(self):
        print(f"Loading washing machine")


class DoLaundry:
    def __init__(self, laundromat):
        self._laundromat = laundromat
        self._drive_to_events = []
        self._load_washing_machine_events = []

    def _drive_to(self, location):
        for drive_event in self._drive_to_events:
            drive_event.drive_to(location)

    def _load_washing_machine(self):
        for washing_machine_event in self._load_washing_machine_events:
            washing_machine_event.load_washing_machine()

    def do_laundry(self):
        self._drive_to(self._laundromat)
        self._load_washing_machine()

在这里，我们看到每个组件都可以连接，而无需了解其消息的接收者。在您的应用程序的主体中，您将处理将每个对象链接到适当的事件方法。一个例子：

drive_to = DriveTo()
load_washing_machine= LoadWashingMachine()
do_laundry = DoLaundry(laundromat="cleaners")

do_laundry.drive_to_events.append(drive_to)
do_laundry.load_washing_machine_events.append(load_washing_machine)

do_laundry.do_laundry()

更进一步，我们可以看到每个事件的逻辑都是多余的。我们可以通过为事件创建一个通用类来简化这一点。每种类型的消息都可以是不同的 observable 实例，接收者可以指定他们订阅的事件类型。

class Observer(ABC):
    @abstractmethod
    def on_next(self, *args, **kwargs):
        pass


class Observable:
    def __init__(self):
        self._observers = []

    def subscribe(self, observer):
        self._observers.append(observer)

    def on_next(self, *args, **kwargs) -> None:
        for observer in self._observers:
            observer.on_next(*args, **kwargs)

重构我们最初的例子：

class DriveTo(Observer):
    def on_next(self, location):
        print(f"Driving To {location}")


class LoadWashingMachine(Observer):
    def on_next(self):
        print(f"Loading washing machine")


class DoLaundry:
    def __init__(self, laundromat):
        self._laundromat = laundromat
        self.drive_observable = Observable()
        self.load_observable = Observable()

    def do_laundry(self):
        self.drive_observable.on_next(self._laundromat)
        self.load_observable.on_next()


do_laundry = DoLaundry(laundromat="cleaners")
do_laundry.drive_observable.subscribe(DriveTo())
do_laundry.load_observable.subscribe(LoadWashingMachine())

do_laundry.do_laundry()

rxpy 库提供了很多这样的实用程序，使您可以非常轻松地编写代码。它可以处理诸如过滤流、组合它们、处理异常、取消订阅事件等等。

Answer 2

使对象更易于组合的另一个选择是使用依赖注入 (DI) 框架或控制反转 (IoC) 框架，例如 pinject。

在这些系统中，我们可以将依赖项定义为普通的构造函数参数，但允许 IoC 系统处理将它们挂钩在一起。像往常一样开始上课：

class Car:
    def drive_to(self, location):
        print(f"Driving To {location}")


class WashingMachine:
    def load(self):
        print(f"Loading washing machine")


class DoLaundry:
    def __init__(self, laundromat, car, washing_machine):
        self._laundromat = laundromat
        self._car = car
        self._washing_machine = washing_machine

    def do_laundry(self):
        self._car.drive_to(self._laundromat)
        self._washing_machine.load()

现在我们可以允许 pinject 将所有内容连接在一起。对于像 consts 这样的东西，您可以使用绑定规范来帮助 pinject 确定值应该是什么：

import pinject


class BindingSpec(pinject.BindingSpec):
    def provide_laundromat(self):
        return "cleaners"


obj_graph = pinject.new_object_graph(binding_specs=[BindingSpec()])
do_laundry = obj_graph.provide(DoLaundry)
do_laundry.do_laundry()