oop-event-bus 0.0.1

A simple OOP event bus

Intro

This package implements an event bus based on the Mediator pattern. It is a powerful tool to avoid coupling and make components work together without any knowledge about each other (loose coupling). It helps:

• to avoid unneeded dependencies;
• to make testing easier;
• to extend the functionality by adding (not changing) the units;
• to reduce the tech debt.

Installation

pip install oop-event-bus

Usage

Create an event class with all the details you need

class UserSignedIn(Event):
    def __init__(self, email: Email):
        self.email = email

Create and register event listeners

class LogSignedInUsersListener(TypedEventListener[UserSignedIn]):
    async def __call__(self, event: UserSignedIn):
        ...

event_bus = EventBus()
event_bus.listen(LogSignedInUsersListener())

Dispatch an event

await event_bus.dispatch(UserSignedIn(user.email))

Why?

Lets imagine the following code:

class SomethingService:
    def __init__(self, logger: Logger, mailer: Mailer, marketing: MarketingService):
        self.logger = logger
        self.mailer = mailer
        self.marketing = marketing

    async def do_something(self):
        self._real()
        self._business()
        self._logic()

        #side effects
        self.logger.info("We did something")

        await self.mailer.send_email("Welcome email")

        #unrelated code that should be executed when we do something
        await self.marketing.increase_marketing_counter()

We have SRP breaks, and a number of side effects unrelated to the business problem. This code is hard to test, it couples SomethingService with the concrete implementations of logging, mailing and marketing services.

With oop-event-bus you can move all side effects and unrelated code out. Also, you will make it possible to extend functionality later without changing the code of this method.

from oop_bus import Event, EventBus

class WeDidSomething(Event):
    ...

class SomethingService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
    async def do_something(self):
        real()
        business()
        logic()

        await self.event_bus.dispatch(WeDidSomething())

You can see that every piece of non-related code was replaced with the single event_bus call here.