I'm interested in making my current Python project conform to the object-oriented paradigm as it was originally conceived by Alan Kay--which, from my understanding, is about objects messaging each-other as much as it's about the objects themselves.
Example code from the sort of package that I'm thinking of may look something like this:
def do_laundry(self):
send(self.car, "DRIVE_TO", (self.laundromat))
washing_machine = self.look_for(WashingMachine)
send(self.washing_machine, "LOAD")
...
I can see how sticking closely to this paradigm could simplify object interfaces, and the basics of message-passing don't seem that difficult to implement, so I'm surprised that I haven't been able to find anything relevant. (All message-related packages that I've found relate to email or other Internet activities.) Is there a package that does what I'm looking for, or am I mistaken in believing that would be a useful thing for Python to have? It's conceivable that is hasn't been done and is potentially useful, but my past experiences indicate that this is unlikely.
P.S. I'm aware of this question, but it has a different focus, it hasn't been active for 10 years, and the link to the Python-specific resource which it refers to is broken.
You could use a pub sub system for this type of work. I'm a big fan of tools like rxpy, but the core concept of Pub-Sub is simple to add yourself. Take this for example:
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()
Here we see that each component can be connected without needing to know much about the receiver of their messages. In the main of your application you would handle linking up each of the objects to the appropriate event methods. An example putting this together:
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()
Going a step further we can see that the logic for each event is redundant. We can simplify this by make a generic class for events. Each type of message can be a different observable instance, and receivers can specify which types of events they subscribe to.
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)
Refactoring our initial example:
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()
The rxpy library offers a LOT of utilities like this, which allow you to compose code very easily. It can handle things like filtering streams, combining them, handling exceptions, unsubscribing to events, and more.