~/Blog

Brandon Rozek

Photo of Brandon Rozek

PhD Student @ RPI studying Automated Reasoning in AI and Linux Enthusiast.

Python Patterns: Subscribe

Published on

Updated on

2 minute reading time

Warning: This post has not been modified for over 2 years. For technical posts, make sure that it is still relevant.

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!
Reply via Email Buy me a Coffee
Was this useful? Feel free to share: Hacker News Reddit Twitter

Published a response to this? :