Django Signals structured as Event emitter

Hi, I don’t know if this is the right title 😆, but I have done some good stuff and I hope that’s could be helpful for you.

While I’m develop the RESTful API for a project (bazaar.ao) that will be available soon, with my team from Ahllia LLC. I have the idea to make this structure (I will show you soon below) for events.

At moment we are developing a monolithic application, but we also want to move for Microservices later, so, with that in mind I decided to keep every monolithic's module (django app) the most independent as possible.

So, to share changes/events within modules I decided to use Django Signals.

The problem

First, I started to create a lot of Signal instances for each event such as:

# in: myproject.user.dispatch

from django.dispatch import Signal

account_created = Signal(['user'])
email_created = Signal(['email'])
phone_created = Signal(['phone'])
# and so on…

And after this, I have imported every single instance on signals file to handle the events.

# in: myproject.user.signals

from django.dispatch import receiver
from myproject.user.dispatch import (
    account_created,
    email_created,
    phone_created
)

@receiver(account_created)
def hdl_account_created(sender, user, **kwargs): pass

@receiver(email_created)
def hdl_email_created(sender, email, **kwargs): pass

@receiver(phone_created)
def hdl_phone_created(sender, phone, **kwargs): pass

Has you can see, this will bring us with a lot of instances and imports, making our application bigger and slow to start.

So I came up with the idea to make things more automated. Using only one instance (which I preferred to call it event) of Singal for each module. All signals (specifically, the keys) will be registered at runtime and only when needed.

First, I'll show you how to use it (it's beautiful ❤️) then I'll show you how I implemented it. You can choose to implement it the same way or use it to do your own implementation.

Using Event

I created a file named events.py in each module that has an instance of the global Event class.

# in: myproject.user.events

# I implemented the global Event class in the core module.
# Which contains all the classes shared between other modules.
from myproject.core.events import Event

EVENTS: dict = {
    'ACCOUNT_CREATED': 'user:account:created',
    'EMAIL_CREATED': 'user:emails:created',
    'PHONE_CREATED': 'user:phones:created'}

event = Event(events=EVENTS)

As you can see, the structure of the event dictionary is quite simple. Where key is the name of the event accessed in the instance and the value is the name of the event sharing between modules or systems. In case the event is being sent to an Event Queue

Due to this fact (events can be shared across systems) it is important to create a naming pattern to avoid duplicates and better grouping. I chose the following pattern:

<app|module|service name>:<context>:<action created|changed|deleted|viewed>

Some thing like: user:account-primary-email:changed

Continuing, I demonstrate below in a view how an event can be emitted:

# in: myproject.user.views

# Just import the event instance
from myproject.user.events import event

def create_user_view(request):
    # ... user creation logic
    user = User.objects.create()
    # emitting the event/signal
    event.send(event.n.ACCOUNT_CREATED, user=user)
    # or 
    event.send(event.name.ACCOUNT_CREATED, user=user)

The attribute n in the event, which can also be accessed via name, contains the name of the event previously defined in the EVENTS dictionary. This name is used as a sender when sending the event/signal.

Now, let's see how to handle the events

# in: myproject.user.signals

# Just import the signal receiver and the event instance
from django.dispatch import receiver
from myproject.user.events import event

@receiver(event, **event.EMAIL_CREATED)
def hlr_email_created(sender, email, **kwargs):
    print('Email Created! ', email)
    # do some quick task
    # Later we can do something like that
    enqueue(
        queue='default',
        event_name=event.n.EMAIL_CREATED,
        data=email)

@receiver(event, **event.ACCOUNT_CREATED)
def hlr_email_created(sender, user, **kwargs):
    print('Account Created', user)

Pay close attention and be careful, in two cases you will access the dictionary key defined in events.py.

First, you need to access the event name (E.g: to send a signal or send the event to the Event Queue).
Second, you need the necessary arguments to connect the function called to execute the event.

In the first case, you access the event name by the attribute n or name of the Event instance: event.n.ACCOUNT_CREATED

In the second case, you access the arguments directly from the instance event.ACCOUNT_CREATED

And that's it, and now let's see how I implemented it.

Event class Implementation

I will show the complete class with some details that I thought it was important to be commented on to facilitate the perception.

# in: myproject.core.events

import uuid
from django.dispatch import Signal

"""
First, I had defined this class to be able to access
dictionary key with dot (event.) in event
name or n attribute.
"""
class DictAsObject:

    defaults: dict

    def __init__(self, defaults: dict) -> None:
        self.defaults = defaults or dict()

    def __getattr__(self, attr: str):
        try:
            value = self.defaults[attr]
        except KeyError:
            raise AttributeError("Invalid key: '%s'" % attr)
        setattr(self, attr, value)
        return value

"""
Then I had define the Event class that inherits
from django.dispatch's Signal class
"""

class Event(Signal):

    __n: DictAsObject

    def __init__(self, providing_args: List[str] = None, use_caching: bool = False, events: dict = dict()) -> None:
        super().__init__(providing_args=providing_args, use_caching=use_caching)
        self.__n = DictAsObject(events)

    def __getattr__(self, attr: str):
        """
        Looks the event name by key in __n DictAsObject 
        instance and cache it on instance
        with the kwargs to django dispatch receiver:
            sender: event name
            dispatch_uid: unique id for event (event name too)
        """
        try:
            event_name: str = getattr(self.__n, attr)
        except AttributeError:
            raise AttributeError(_("the '%s' event was not registered") % attr)
        else:
            kwargs = dict(sender=event_name, dispatch_uid=str(uuid.uuid4())))
            setattr(self, attr, kwargs)
        return kwargs

    @property
    def n(self):
        return self.__n

    @property
    def name(self):
        return self.__n

And that's it, I hope you enjoyed it and that it's useful for you. If you have any criticisms or suggestions, don't hesitate to leave them in the comments. Maybe some disadvantage or advantage that I haven't mentioned...

If it was helpful or if you found it interesting, also leave an 🙋🏽‍♂️❤️ in the comments. Thank you, looking forward to the next article. Enjoy and follow my work.