Skip to content

Event Bus Base

event_bus_base

Event bus base class compliant with the application EventBusPort contract.

Defines the EventBusBase abstract interface for in-process message dispatch
with separate policies for events (multi-handler fan-out) and commands
(single-handler routing).

EventBusBase

Bases: ABC

Base class for event buses.

Implementations handle
  • Publishing events to one or more registered handlers.
  • Sending commands to a single registered handler.
  • Registering handlers for specific message types.
Example 
class OrderCreated:
    def __init__(self, order_id: str) -> None:
        self.order_id = order_id


class InMemoryEventBus(EventBusBase[dict[str, object], dict[str, object], object]):
    def register_handler(self, message_type, handler) -> None: ...
    async def publish(self, event) -> Result[None, EventBusError]: ...
    async def send(self, command) -> Result[None, EventBusError]: ...


async def order_created_handler(event: OrderCreated) -> None:
    print(f"Processing order {event.order_id}")


bus = InMemoryEventBus()
bus.register_handler(OrderCreated, order_created_handler)
await bus.publish(OrderCreated(order_id="42"))
Source code in src/forging_blocks/infrastructure/event_buses/event_bus_base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class EventBusBase[EventPayloadType, CommandPayloadType, HandlerType](ABC):
    """Base class for event buses.

    Implementations handle:
      - Publishing events to one or more registered handlers.
      - Sending commands to a single registered handler.
      - Registering handlers for specific message types.

    Example:
        ```python
        class OrderCreated:
            def __init__(self, order_id: str) -> None:
                self.order_id = order_id


        class InMemoryEventBus(EventBusBase[dict[str, object], dict[str, object], object]):
            def register_handler(self, message_type, handler) -> None: ...
            async def publish(self, event) -> Result[None, EventBusError]: ...
            async def send(self, command) -> Result[None, EventBusError]: ...


        async def order_created_handler(event: OrderCreated) -> None:
            print(f"Processing order {event.order_id}")


        bus = InMemoryEventBus()
        bus.register_handler(OrderCreated, order_created_handler)
        await bus.publish(OrderCreated(order_id="42"))
        ```
    """

    @abstractmethod
    async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
        """Publish a domain event to all registered handlers.

        Args:
            event: The domain event to publish.

        Returns:
            A ``Result`` indicating success or an ``EventBusError``.

        """

    @abstractmethod
    async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
        """Send a command to its registered handler.

        Args:
            command: The command to dispatch.

        Returns:
            A ``Result`` indicating success or an ``EventBusError``.

        """

    @abstractmethod
    def register_handler(
        self,
        message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]],
        handler: HandlerType,
    ) -> None:
        """Register a handler for the given message type.

        Args:
            message_type: The message class to handle.
            handler: A handler instance.

        """

publish(event: Event[EventPayloadType]) -> Result[None, EventBusError] abstractmethod async

Publish a domain event to all registered handlers.

Parameters:

Name Type Description Default
event Event[EventPayloadType]

The domain event to publish.

 required

Returns:

Type Description
Result[None, EventBusError]

A Result indicating success or an EventBusError.
Source code in src/forging_blocks/infrastructure/event_buses/event_bus_base.py 
47
48
49
50
51
52
53
54
55
56
57
@abstractmethod
async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
    """Publish a domain event to all registered handlers.

    Args:
        event: The domain event to publish.

    Returns:
        A ``Result`` indicating success or an ``EventBusError``.

    """

send(command: Command[CommandPayloadType]) -> Result[None, EventBusError] abstractmethod async

Send a command to its registered handler.

Parameters:

Name Type Description Default
command Command[CommandPayloadType]

The command to dispatch.

 required

Returns:

Type Description
Result[None, EventBusError]

A Result indicating success or an EventBusError.
Source code in src/forging_blocks/infrastructure/event_buses/event_bus_base.py 
59
60
61
62
63
64
65
66
67
68
69
@abstractmethod
async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
    """Send a command to its registered handler.

    Args:
        command: The command to dispatch.

    Returns:
        A ``Result`` indicating success or an ``EventBusError``.

    """

register_handler(message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]], handler: HandlerType) -> None abstractmethod

Register a handler for the given message type.

Parameters:

Name Type Description Default
message_type type[Event[EventPayloadType]] | type[Command[CommandPayloadType]]

The message class to handle.

 required
handler HandlerType

A handler instance.

 required
Source code in src/forging_blocks/infrastructure/event_buses/event_bus_base.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
@abstractmethod
def register_handler(
    self,
    message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]],
    handler: HandlerType,
) -> None:
    """Register a handler for the given message type.

    Args:
        message_type: The message class to handle.
        handler: A handler instance.

    """