Skip to content

In Memory Event Bus

in_memory_event_bus

In-memory implementation of the EventBusPort port.

Dispatches events to multiple registered handlers (fan-out) and
commands to a single registered handler. Handlers are looked up
by the exact type of the message.

InMemoryEventBus

Bases: EventBusPort[EventPayloadType, CommandPayloadType, HandlerType]

In-memory event bus with separate event/command dispatch.

Attributes:

Name Type Description
_event_handlers dict[type[Event[EventPayloadType]], list[_Handler[Event[EventPayloadType]]]]

Per-event-type list of handlers.
_command_handlers dict[type[Command[CommandPayloadType]], _Handler[Command[CommandPayloadType]]]

Per-command-type single handler.
Example 
class StubEvent[T]:
    def __init__(self) -> None:
        pass


class OrderCompleted(StubEvent[dict[str, object]]):
    def __init__(self, order_id: str) -> None:
        self.order_id = order_id


class OrderCompletedHandler:
    async def handle(self, event: OrderCompleted) -> None:
        print(f"Order completed: {event.order_id}")


bus = InMemoryEventBus[dict[str, object], object, object]()
bus.register_handler(OrderCompleted, OrderCompletedHandler())
await bus.publish(OrderCompleted(order_id="abc-123"))
Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus.py 
 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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class InMemoryEventBus[EventPayloadType, CommandPayloadType, HandlerType](
    EventBusPort[EventPayloadType, CommandPayloadType, HandlerType]
):
    """In-memory event bus with separate event/command dispatch.

    Attributes:
        _event_handlers: Per-event-type list of handlers.
        _command_handlers: Per-command-type single handler.

    Example:
        ```python
        class StubEvent[T]:
            def __init__(self) -> None:
                pass


        class OrderCompleted(StubEvent[dict[str, object]]):
            def __init__(self, order_id: str) -> None:
                self.order_id = order_id


        class OrderCompletedHandler:
            async def handle(self, event: OrderCompleted) -> None:
                print(f"Order completed: {event.order_id}")


        bus = InMemoryEventBus[dict[str, object], object, object]()
        bus.register_handler(OrderCompleted, OrderCompletedHandler())
        await bus.publish(OrderCompleted(order_id="abc-123"))
        ```
    """

    __slots__ = ("_command_handlers", "_event_handlers")

    def __init__(self) -> None:
        self._event_handlers: dict[
            type[Event[EventPayloadType]], list[_Handler[Event[EventPayloadType]]]
        ] = {}
        self._command_handlers: dict[
            type[Command[CommandPayloadType]], _Handler[Command[CommandPayloadType]]
        ] = {}

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

        For event types, multiple handlers can be registered (fan-out).
        For command types, only one handler is allowed per type.

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

        """
        if issubclass(message_type, Event):
            self._event_handlers.setdefault(message_type, []).append(
                cast(_Handler[Event[EventPayloadType]], handler)
            )
            return
        self._command_handlers[message_type] = cast(_Handler[Command[CommandPayloadType]], handler)

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

        Args:
            event: The domain event.

        Returns:
            ``Ok(None)`` on success, or ``Err(EventBusError)`` if any
            handler raises.

        """
        handlers = self._event_handlers.get(type(event), [])
        for handler in handlers:
            try:
                await handler.handle(event)
            except Exception as exc:
                return Err(EventBusError(str(exc)))
        return Ok(None)

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

        Args:
            command: The command.

        Returns:
            ``Ok(None)`` on success, or ``Err(EventBusError)`` if the
            handler raises or no handler is registered.

        """
        handler = self._command_handlers.get(type(command))
        if handler is None:
            return Err(EventBusError(f"No handler registered for {type(command).__name__}"))
        try:
            await handler.handle(command)
        except Exception as exc:
            return Err(EventBusError(str(exc)))
        return Ok(None)

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

Register a handler for a message type.

For event types, multiple handlers can be registered (fan-out).
For command types, only one handler is allowed per 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/in_memory_event_bus.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def register_handler(
    self,
    message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]],
    handler: HandlerType,
) -> None:
    """Register a handler for a message type.

    For event types, multiple handlers can be registered (fan-out).
    For command types, only one handler is allowed per type.

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

    """
    if issubclass(message_type, Event):
        self._event_handlers.setdefault(message_type, []).append(
            cast(_Handler[Event[EventPayloadType]], handler)
        )
        return
    self._command_handlers[message_type] = cast(_Handler[Command[CommandPayloadType]], handler)

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

Publish an event to all registered handlers.

Parameters:

Name Type Description Default
event Event[EventPayloadType]

The domain event.

 required

Returns:

Type Description
Result[None, EventBusError]

Ok(None) on success, or Err(EventBusError) if any
Result[None, EventBusError]

handler raises.
Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
    """Publish an event to all registered handlers.

    Args:
        event: The domain event.

    Returns:
        ``Ok(None)`` on success, or ``Err(EventBusError)`` if any
        handler raises.

    """
    handlers = self._event_handlers.get(type(event), [])
    for handler in handlers:
        try:
            await handler.handle(event)
        except Exception as exc:
            return Err(EventBusError(str(exc)))
    return Ok(None)

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

Send a command to its registered handler.

Parameters:

Name Type Description Default
command Command[CommandPayloadType]

The command.

 required

Returns:

Type Description
Result[None, EventBusError]

Ok(None) on success, or Err(EventBusError) if the
Result[None, EventBusError]

handler raises or no handler is registered.
Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
    """Send a command to its registered handler.

    Args:
        command: The command.

    Returns:
        ``Ok(None)`` on success, or ``Err(EventBusError)`` if the
        handler raises or no handler is registered.

    """
    handler = self._command_handlers.get(type(command))
    if handler is None:
        return Err(EventBusError(f"No handler registered for {type(command).__name__}"))
    try:
        await handler.handle(command)
    except Exception as exc:
        return Err(EventBusError(str(exc)))
    return Ok(None)