website/content/blog/pysubscribepattern.md

title	date	draft	tags	medium_enabled
Python Patterns: Subscribe	2020-04-14T07:53:46-04:00	false	Python	true

It is common for larger applications to have modules that publishes and subscribes to events. This post will outline a couple ways to achieve this using decorators and standard methods.

Single Event

First let us concern ourselves with a single event since that's the easiest. Here we will create an application class that stores callbacks of functions through the subscribe decorator. Calling emit will send a message to all the functions stored in self.callbacks.

from typing import Callable, List
class Application:
    def __init__(self):
        self.callbacks: List[Callable] = []
    def subscribe(self, func: Callable):
        if not callable(func):
            raise ValueError("Argument func must be callable.")
        self.callbacks.append(func)
        return func
    def emit(self, message):
        for callback in self.callbacks:
            callback(message)

Here is an example of its usage:

app = Application()

@app.subscribe
def test1(message):
    print("Function 1:", message)

def test2(message):
    print("Function 2:", message)

app.subscribe(test2)

app.emit('Hello World')

Function 1: Hello World
Function 2: Hello World

Multiple Events

Let's say you want the application to handle different types of events. Now self.callbacks is a dictionary of lists, where the key is the event and the list is the same as the last section. There's an additional layered function on top of subscribe this time in order to handle passing an argument into the decorator.

from collections import defaultdict
from typing import Callable, Optional
class Application:
    def __init__(self):
        self.callbacks: Dict[str, List[Callable]] = defaultdict(list)
    def on(self, event: str, func: Optional[Callable] = None):
        def subscribe(func: Callable):
            if not callable(func):
                raise ValueError("Argument func must be callable.")
            self.callbacks[event].append(func)
            return func
        if func is None:
            return subscribe
        subscribe(func)
    def emit(self, event, message):
        for callback in self.callbacks[event]:
            callback(message)

To show its usage lets first create an instance of Application

app = Application()

Now let's subscribe a couple functions to event1

@app.on('event1')
def test1(message):
    print("Function 1:", message)

def test3(message):
    print("Function 3:", message)

app.on('event1', test3)

Now to subscribe a couple events to event2

# Subscribed to event 2
@app.on('event2')
def test2(message):
    print("Function 2:", message)

def test4(message):
    print("Function 4:", message)

app.on('event2', test4)

We can also subscribe to both events

# Subscribed to both events
@app.on('event1')
@app.on('event2')
def test5(message):
    print("Function 5:", message)

app.emit('event1', 'Hello, World!')

Function 1: Hello, World!
Function 3: Hello, World!
Function 5: Hello, World!

app.emit('event2', 'Goodbye, World!')

Function 2: Goodbye, World!
Function 4: Goodbye, World!
Function 5: Goodbye, World!