DEV Community

loading...

Build your own event system in Python

kuba_szw profile image Kuba ・3 min read

Event system is not hard to build on your own. There are plenty of libraries ready to use, but for better understanding I want to implement it by myself.

The idea of this post comes from this video. Today I was searching the internet for tips how to manage my project which has a lot of different modules and I wanted to do it nice and clean. One of the ideas I found is Observer pattern from “Gang of four” book. This should be very simple implementation of it.

First thing is that you have subscribers. They subscribe to different types of events. Every time the event is posted, you have to notify the subscriber about it.

First create two methods for subscribing the event, and for posting.

# src/event.py

from collections import defaultdict

subscribers = defaultdict(list)

def subscribe(event_type, fn):
    subscribers[event_type].append(fn)

def post_event(event_type, data): 
    if event_type in subscribers:
        for fn in subscribers[event_type]:
            fn(data)
Enter fullscreen mode Exit fullscreen mode

In this example I will be creating the new user which will invoke event new_user_created. Subscriber for this event will be module for sending welcome message by email.

Let's create a user. It will be simple dictionary, but it can be an object too.

# src/user.py

from .event import post_event

def register_new_user(name, password, email ):
    user = dict( name = name, 
                password = password, 
                email = email)

    post_event("new_user_created", user)

Enter fullscreen mode Exit fullscreen mode

Now handle sending emails. We will need two things here. Some kind of email provider and some kind of handler for it. I want my email provider class to handle all 'business logic' only, so I created modules folder for such.

# src/modules/email.py 

class Email:
    def sendEmail( email, subject, message ):
        print("==========================")
        print(f"From: {email}")
        print(f"Subject: {subject}")
        print(message)
        print("==========================")
Enter fullscreen mode Exit fullscreen mode

And handler for email.

# src/email_handler.py

from .modules.email import Email 
from .event import subscribe

def handle_user_registered_event(user):
    Email.sendEmail(user['email'], 'Welcome!', 'Some welcome message')

def setup_email_event_handlers():
    subscribe('new_user_created', handle_user_registered_event)
Enter fullscreen mode Exit fullscreen mode

Now let's connect everything together. Make an app file.

# src/app.py

from src.user import register_new_user
from src.email_handler import setup_email_event_handlers

setup_email_event_handlers()

register_new_user('Jakub', 'secret', 'name@domain.com')

Enter fullscreen mode Exit fullscreen mode

First thing to do is to set email handler events up. That means, subscribe to event new_user_created.

If all the subscriptions are done, register new user. Output should be as follows.

==========================
From: name@domain.com
Subject: Welcome!
Some welcome message
==========================
Enter fullscreen mode Exit fullscreen mode

Extend the system

What if we want to extend this. Let's add some database. In modules, add database.py file and sample implementation of database as a list of users.

# src/modules/database.py 

class DB:
    users = [ ]
    def register_new_user( user ):
        DB.users.append(user)
        print('=======================')
        print(DB.users)
        print('=======================')

Enter fullscreen mode Exit fullscreen mode

And handler for it.

# src/database_handler.py

from .modules.database import DB
from .event import subscribe

def handle_user_registered_event(user):
    DB.register_new_user(user)

def setup_database_event_handlers():
    subscribe('new_user_created', handle_user_registered_event)
Enter fullscreen mode Exit fullscreen mode

Now the only thing to do is set database handler up in app file.

# src/app.py

from src.user import register_new_user
from src.email_handler import setup_email_event_handlers
from src.database_handler import setup_database_event_handlers 

setup_email_event_handlers()
setup_database_event_handlers()

register_new_user('Jakub', 'secret', 'name@domain.com')
Enter fullscreen mode Exit fullscreen mode

Now the result should be:

==========================
From: name@domain.com
Subject: Welcome!
Some welcome message
==========================
==========================
[{'name': 'Jakub', 'password': 'secret', 'email': 'name@domain.com'}]
==========================
Enter fullscreen mode Exit fullscreen mode

Nice thing about it, is that this way keeps your modules independent of each other and keeps their cohesion strong. Besides that, it allows you to manipulate order of listeners in event very quickly. No more 'ctrl + c | ctrl + v' whole code 😉.

Comment what do you think about it! How do you build workflow in your apps between multiple modules?

Discussion (0)

pic
Editor guide