是否有支持对象消息传递的Python包?

Question

我对使我当前的Python项目符合Alan Kay最初设想的面向对象范式很感兴趣——根据我的理解，它是关于对象之间的消息传递，就像它是关于对象本身一样。

我正在考虑的包的示例代码可能看起来像这样:

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

我可以看到紧密遵循这个范例可以简化对象接口，并且消息传递的基础似乎并不难实现，所以我很惊讶我没有找到任何相关的东西。(我找到的所有与信息相关的软件包都与电子邮件或其他互联网活动有关。)是否有一个包可以做我正在寻找的东西，或者我错误地认为这对Python来说是一个有用的东西?可以想象，还没有人这样做，而且可能有用，但我过去的经验表明，这是不可能的。

注:我知道这个问题，但它有一个不同的焦点，它已经不活跃了10年，它所引用的特定于python的资源的链接被破坏了。

Answer 1

对于这类工作，您可以使用发布子系统。我是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()

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

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将所有内容连接在一起。对于像const这样的东西，你可以使用绑定规范来帮助确定应该是什么值:

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()