Usage Example

Create your commands, receivers, events, and handlers:

yourapp/commands/add_item_to_cart.py:

from attrs import define, field

from bus_ride import Command, Receiver, ReceiverReturnType, ReturnMessage


@define
class AddItemToCart(Command):
    cart_id: int = field()
    item_id: int = field()
    qty: int = field()


@define
class AddItemToCartReceiver(Receiver):
    command: AddItemToCart = field()

    handles_command_cls = AddItemToCart

    def execute_command(self) -> ReceiverReturnType:
        from ..events.item_added_to_cart import ItemAddedToCart

        cart_item_id = self._save_item_to_cart()
        return [
            ReturnMessage(data=cart_item_id),
            ItemAddedToCart(cart_item_id=cart_item_id),
        ]

    def _save_item_to_cart(self) -> int:
        # pretend we actually saved it
        return 1

yourapp/events/item_added_to_cart.py:

from attrs import define, field

from bus_ride import Event, Handler, HandlerReturnType


@define
class ItemAddedToCart(Event):
    cart_item_id: int = field()


class ItemAddedToCartHandler(Handler):
    event: ItemAddedToCart = field()

    handles_events = [ItemAddedToCart]

    def handle_event(self) -> HandlerReturnType:
        self._notify_sales_team()
        return []

    def _notify_sales_team(self) -> None:
        # obviously you would do something real here:
        print(f"We got one! {self.event}")

yourapp/views.py:

from bus_ride import MessageBus
from .commands import AddItemToCart


def add_item_to_cart_view(request):
    command = AddItemToCart(
        cart_id=request.session["cart_id"],
        item_id=request.POST.get("item_id"),
        qty=request.POST.get("quantity", 1),
    )
    bus = MessageBus(command)
    result = bus()
    cart_item_id = result[0].data
    return {"result": "OK", "cart_item_id": cart_item_id}

Note

We recommend having unit tests which use mock.patch to make sure your receivers and handlers are being called as expected. Please see the unit tests of this library for examples.