Hi 😀

In this blog we are going to build our own python library “Signals”

What is this library about?

This is a very basic publisher/subscriber implementation where a publisher publishes event (emit) and its subscribers react on those events. somewhat similar to when someone calls your name and you react with “ha?” here someone is “publisher”, your name is “event”, you are subscriber of the event and “ha?” is the action you performed.

Code

to build this is python I am going to follow singleton approach which means only 1 instance is created of a class. this can be done as below.

from typing import Union

class Signal:
    instance: Union['Signal', None] = None

    def __new__(cls):
        if cls.instance is None:
            cls.instance = super().__new__(cls)
        return cls.instance

This is an example of singleton class because of if condition, it basically checks if the instance is already existing this will return same instance. else it will create a new instance.

Now comes the initialization of Signal class

def __init__(self, name):
    if not self.is_initialized:
        self.signal_name = name.lower()
        self.listners.setdefault(self.signal_name, [])
        self.is_initialized = True

similar to __new__ function __init__ also runs just once. and in this function, we name to our signal.
with this we can do.

joker = Signal("rob")
face2 = Signal("rob") # Also know as 2face

here we have created 2 instances of same signal the idea behind above like is 2 persons are planning to do robbery.
Now let create a subscriber function of Signal class that will register the action that needs to be performed based of event.

@classmethod
def sub(cls, signal_name):
    signal_name = signal_name.lower()
    @wrapers(cls.sub)
    def wrapper(callback):
        cls.listners[signal_name].append(callback)
        return callback
    return wrapper

This is the simple decorator that we create in python but instead of running the callback function we are storing it. we’ll run when we get the event (emit)
below is the action that we’ll perform for “rob” event

@Signal.sub("rob")
def batman():
    return "batman is here to stop robbery"

Below is emit function of our Signal class.

def emit(self, *args, **kwargs):
    if not self.signal_name in self.listners:
        return
    for callback in self.listners[self.signal_name]:
        callback(*args, **kwargs)

Thats it we have our custom Signal library ready. to test this, we can run

joker.emit()
face2.emit()

output would be “batman is here to stop robbery”.

Final code.

from functools import wraps
from collections import defaultdict
from typing import Callable, Union

class Signal:

    instance: Union['Signal', None] = None
    is_initialized = False
    listners: defaultdict = defaultdict(list)

    def __new__(cls, name):
        if cls.instance is None:
            cls.instance = super().__new__(cls)
        return cls.instance

    def __init__(self, name):
        if not self.is_initialized:
            self.signal_name = name.lower()
            self.listners.setdefault(self.signal_name, [])
            self.is_initialized = True

    @classmethod
    def sub(cls, signal_name):
        """Subscribe to a signal."""
        signal_name = signal_name.lower()
        @wraps(cls.sub)
        def wrapper(callback):
            cls.listners[signal_name].append(callback)
            return callback
        return wrapper

    def emit(self, *args, **kwargs):
        """Emit a signal."""
        if not self.signal_name in self.listners:
            return
        for callback in self.listners[self.signal_name]:
            callback(*args, **kwargs)

>>> from main import Signal
>>> joker = Signal("rob")
>>> face2 = Signal("rob")
>>> @Signal.sub("rob")
... def batman():
...     print("batman is here to stop robbery")
...
>>> joker.emit()
batman is here to stop robbery
>>> face2.emit()
batman is here to stop robbery

Let me know if you enjoyed this 10min project :)

Thanks for reading.