Using Python types for fun and profit: the Mediator pattern

Introduction

The mediator pattern is a pattern used when you want to simplify communication i.e. message dispatching in complex applications. It involves building a mediator where each objects can delegate its communication, and which routes the messages to the right receiver(s).

It looks like this:

A short explanation of this diagram:

1. First there is the Mediator interface, which defines the message which can be routed through this Mediator.
2. The ConcreteMediator implements the Mediator interface, and coordinates the communication between the ConcreteWorker objects. It knows about the ConcreteWorker objects and how they should communicate
3. The Worker is an interface (or in some languages an abstract class) which is an interface for the communication between the ConcreteWorkers
4. The ConcreteWorker(x) classes implement the Worker interface, and communicate with other ConcreteWorker classes through the Mediator.

Implementation in Python

Because we need some types before we define them, we need this preliminary:

from __future__ import annotations

You can find the docs about this import here.

The Event class

Since our workers will be sending events to each other, we will start by defining our Event class:

class Event:
    _payload: str = None

    def __init__(self, initial_payload: str):
        self._payload = initial_payload

    def __str__(self) -> str:
        return f"Event payload: {self._payload}"

To keep our example simple, our events have just a string payload.

The Mediator base class

Now we are ready to define the Mediator base class:

class Mediator:
    def notify(self, sender: Worker, event: Event):
        pass

    def add_worker(self, worker):
        pass

Some notes:

1. The notify() method is used to notify, i.e. alert, all the workers connected to this mediator of an event. Also note that the sender parameter is of type Worker since those are objects raising the events
2. In add_worker() we connect a worker to the mediator

The Worker base class

This is the base class for the Worker, and has send() and receive() methods to handle sending and receiving messages and events. Note that a Worker does not know where his messages are going.

In our simplified case the Worker also doesn’t know where a message or event is coming from. Had we had a more advanced setup, the message- or event-objects could have contained some notion of a sender:

class Worker:
    def set_mediator(self, new_mediator: Mediator):
        pass

    def send(self, new_event: Event):
        pass

    def receive(self, new_event: Event):
        pass

The concrete classes

We will start with ConcreteMediator class:

class ConcreteMediator(Mediator):
    _workers: list[Worker] = None

    def __init__(self):
        self._workers = []

    def add_worker(self, worker: Worker):
        self._workers.append(worker)

    def notify(self, sender, new_event: Event):
        for worker in self._workers:
            if worker != sender:
                worker.receive(event)

Some notes:

1. The Mediatorkeeps a list of its workers in the list _workers
2. The addWorker() method simply adds a worker to this list
3. In the notify() method, we iterate over all the registered workers, making sure that we exclude the original sender from receiving the event so we do not get infinite loops.

Now we can have a look at the ConcreteWorker class:

class ConcreteWorker(Worker):
    _name: str = None
    _mediator: Mediator = None

    def __init__(self, name: str):
        self._name = name

    def set_mediator(self, new_mediator: Mediator):
        self._mediator = new_mediator

    def send(self, new_event: Event):
        self._mediator.notify(self, event)

    def receive(self, new_event: Event):
        print(f"{self._name} received {event}")

Some explanation is needed here:

1. Each worker has a name and associated mediator. In a more advanced example we could have more mediators per worker, but to keep it simple we will just keep it to one.
2. setMediator() just sets the mediator
3. The send() method prints out the name of event to be sent, the notifies the mediator, passing the worker and the event to the notify() method
4. receive() simply receives an event and prints it out
5. Note that for clarity both the _name and the _mediator instance variables have type-annotations

Putting it together

We can now test this code:

if __name__ == "__main__":
    mediator: Mediator = ConcreteMediator()

    worker1 = ConcreteWorker("worker1")
    worker2 = ConcreteWorker("worker2")
    worker3 = ConcreteWorker("worker3")

    mediator.add_worker(worker1)
    mediator.add_worker(worker2)
    mediator.add_worker(worker3)

    worker1.set_mediator(mediator)
    worker2.set_mediator(mediator)
    worker3.set_mediator(mediator)

    event: Event = Event("hello")

    worker1.send(event)
    worker2.send(event)
    worker3.send(event)

A short explanation:

• We construct a ConcreteMediator and three ConcreteWorkers
• We add the workers to the mediator
• And set the workers’ mediator to the ConceteMediator objhect.
• Then we start sending messages.
• Note that the mediator variable is typed as Mediator and not ConcreteMediator, so we could easily swap out ConcreteMediator for any other class subclassing Mediator

Save this file and try it out. You should see something like this:

worker2 received Event payload: hello
worker3 received Event payload: hello
worker1 received Event payload: hello
worker3 received Event payload: hello
worker1 received Event payload: hello
worker2 received Event payload: hello

Conclusion

Implementing this design pattern was very easy in Python. The only doubt I have is the lack of strong typing which can lead to subtle errors. On the other hand, using type-hints wisely and the use of linters can greatly improve the quality and the stability of the code.

Also the __future__ import also greatly helps by making the purpose and intent of each piece of code much clearer.