A simple Python 3.5+ event bus.

A way to trigger multiple subsequent functions.

pip3 install event-bus

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

fromevent_busimportEventBusbus=EventBus() @bus.on('hello')defsubscribed_event(): print('World!') defsome_func(): print('Hello') bus.emit('hello') >>>some_func() "Hello""World!"

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.

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.

Here are some examples on real world usage.

# Mock Database. USERS={1:{'name': 'Ricky Bobby', 'email': 'someuser@gmail.com', } } @bus.on('new:user')defsend_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')defsend_temporary_pass(user_id): user=USERS.get(user_id) # Logic for sending temp pass email...print('Sent temp pass email to{}'.format(user['name'])) defcreate_user(): # Logic for creating a user...user_id=1bus.emit('new:user', user_id) >>>create_user() 'Sent welcome email to Ricky Bobby''Sent temp pass email to Ricky Bobby'

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')defupdate_avg_ratings(): # Update avg ratings in DB...print("Finished updating ratings.") @bus.emit_after('update:ratings:avg')defadd_rating(): # Creating a new rating...print("Added new rating.") >>>add_rating() "Added new rating.""Finished updating ratings."

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')defevent_one(param): globalGLOBAL_VARGLOBAL_VAR=param@bus.on('event')defevent_two(param): globalGLOBAL_VARGLOBAL_VAR="I don't get called."defsome_func(): bus.emit_only('event', 'event_one', 'it works!') >>>some_func() >>>print(GLOBAL_VAR) 'it works!'

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.

fromevent_bus.exceptionsimportEventDoesntExist@bus.on('fake_event')defevent_one(): passdefsome_func(): try: bus.remove_event('event_one', 'fake_event') exceptEventDoesntExist: # Handle error here..passelse: print("Removed event.") >>>bus.event_count1>>>some_func() "Removed event.">>>bus.event_count0