Event Bus

A simple Python 3.5+ event bus.

Purpose

A way to trigger multiple subsequent functions.

Installation

pip3 install event-bus

Usage

The EventBus is meant to be a singleton used throughout an application.

from event_bus import EventBus

bus = EventBus()

@bus.on('hello')
def subscribed_event():
    print('World!')

def some_func():
    print('Hello')
    bus.emit('hello')

>>> some_func()
"Hello"
"World!"

Performance.

After building the library I started to think about performance and created tests for multiple scenarios. You can learn exactly what each test is doing by looking at the performance_testing.py file inside the tests/ directory.

Below are some metrics under 3 different scenarios:

• CPU Heavy(Fibonacci sequence.)
• IO Heavy(30 File reads, loading json.)
• Network Heavy(100 GET requests to a website.)

Because of the results of the tests I decided to add threading to the library. passing threads=True in the emit(event, *args, **kwargs) method will run the code using multi-threading, this can significantly speed up the events.

Design choices

In some of the methods I require passing in a string for the func_name parameter.

I decided to do this to not require users to import the subscribed events into the file.

In that case it would've been better to just call the functions if they're already imported.

Real world usage

Here are some examples on real world usage.

# Mock Database. 
USERS = {
    1: {
        'name': 'Ricky Bobby',
        'email': '[email protected]',
    }
 }


@bus.on('new:user')
def send_welcome_email(user_id):
     user = USERS.get(user_id)

     # Logic for sending email...
     print('Sent welcome email to {}'.format(user['name']))

@bus.on('new:user')
def send_temporary_pass(user_id):
    user = USERS.get(user_id)
    
    # Logic for sending temp pass email...
    print('Sent temp pass email to {}'.format(user['name']))

def create_user():
    # Logic for creating a user...
    user_id = 1
    bus.emit('new:user', user_id)

>>> create_user()
'Sent welcome email to Ricky Bobby'
'Sent temp pass email to Ricky Bobby'

Emitting events after

There is a decorator for emitting events after code completion.

This is great for functions that are standalone.

Note: This way doesnt allow the passing of args and kwargs into the events.

@bus.on('update:ratings:avg')
def update_avg_ratings():
    # Update avg ratings in DB...
    print("Finished updating ratings.")

@bus.emit_after('update:ratings:avg')
def add_rating():
    # Creating a new rating...
    print("Added new rating.")
    
>>> add_rating()
"Added new rating."
"Finished updating ratings."

Emitting specific events

There might be times when you don't want to emit all the functions that are subscribed to an event.

The emit_only(event: str, func_names: Union[str, List[str]], *args, **kwargs) method allows this.

The code below is an example.

GLOBAL_VAR = 'var_1'

@bus.on('event')
def event_one(param):
    global GLOBAL_VAR
    GLOBAL_VAR = param


@bus.on('event')
def event_two(param):
    global GLOBAL_VAR
    GLOBAL_VAR = "I don't get called."


def some_func():
    bus.emit_only('event', 'event_one', 'it works!')

>>> some_func()
>>> print(GLOBAL_VAR)
'it works!'

Removing subscribed events.

For some reason you might want to completely remove a subscribed event.

This can be achieved with the method remove_event(event: str, func_name: str)

Note: This can also raise a EventDoesntExist exception.

from event_bus.exceptions import EventDoesntExist

@bus.on('fake_event')
def event_one():
    pass

def some_func():
    try:
        bus.remove_event('event_one', 'fake_event')
    except EventDoesntExist:
        # Handle error here..
        pass
    else:
        print("Removed event.")

>>> bus.event_count
1
>>> some_func()
"Removed event."
>>> bus.event_count
0